rltbl/

core.rs

//! # rltbl/relatable
//!
//! This is [relatable](crate) (rltbl::[core](crate::core)).

use crate::{self as rltbl};
use rltbl::{
    git,
    select::{Select, SelectField},
    sql::{
        self, CachingStrategy, DbActiveConnection, DbConnection, DbKind, DbTransaction, JsonRow,
        MemoryCacheKey, SqlParam, VecInto as _,
    },
    table::{Cell, Column, Datatype, Message, Row, Table},
};

use anyhow::Result;
use colored::Colorize;
use csv::{QuoteStyle, ReaderBuilder, Writer, WriterBuilder};
use indexmap::IndexMap;
use lazy_static::lazy_static;
use minijinja::{path_loader, Environment};
use rand::{rngs::StdRng, seq::IteratorRandom as _, Rng as _, SeedableRng as _};
use regex::Regex;
use serde::{Deserialize, Serialize};
use serde_json::{json, to_value, Value as JsonValue};
use sprintf::sprintf;
use std::{
    collections::{HashMap, HashSet},
    fmt::Display,
    fs::File,
    io::Write,
    path::Path as FilePath,
    str::FromStr,
    sync::Mutex,
};
use tabwriter::TabWriter;

/// Default location of the [relatable](crate) database
pub static RLTBL_DEFAULT_DB: &str = ".relatable/relatable.db";

/// Used to calculate the _order field when a new row is added to a table that has metacolumns
pub static NEW_ORDER_MULTIPLIER: usize = 1000;

// The maximum length of the list of previously (un)done commands to fetch when retrieving a user's
// history.
pub static HISTORY_MAX: usize = 1000;

/// The default limit on the number of rows to return in a fetch.
pub static DEFAULT_LIMIT: usize = 100;

/// THe maximum number of rows to return in a fetch.
pub static MAX_LIMIT: usize = 1000;

lazy_static! {
    pub static ref CACHE: Mutex<HashMap<MemoryCacheKey, Vec<JsonRow>>> = Mutex::new(HashMap::new());
}

/// Various errors generated by [relatable](crate)
#[derive(Debug)]
pub enum RelatableError {
    /// An error in the configuration of a ChangeSet:
    ChangeError(String),
    /// An error in the [relatable](crate) configuration:
    ConfigError(String),
    // /// An error that occurred while reading or writing to a CSV/TSV:
    // CsvError(csv::Error),
    /// An error involving the data:
    DataError(String),
    // /// An error generated by the underlying database:
    // DatabaseError(sqlx::Error),
    /// An error that occurred while interacting with git
    GitError(String),
    /// An error generated when the database is missing
    InitError(String),
    /// An error from an unsupported format
    FormatError(String),
    /// An error in the inputs to a function:
    InputError(String),
    /// An error that occurred while reading/writing to stdio:
    IOError(std::io::Error),
    /// An error when a record cannot be found.
    MissingError(String),
    /// An error that occurred while serialising or deserialising to/from JSON:
    SerdeJsonError(serde_json::Error),
    /// An error that occurred while parsing a regex:
    RegexError(regex::Error),
    /// An error when a table cannot be found.
    TableError(String),
    /// An error that occurred because of a user's action
    UserError(String),
}

impl Display for RelatableError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{:?}", self)
    }
}

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

/// The main [rltbl](crate) struct.
#[derive(Debug)]
pub struct Relatable {
    pub root: String,
    pub readonly: bool,
    pub connection: DbConnection,
    // pub minijinja: Environment<'static>,
    pub default_limit: usize,
    pub max_limit: usize,
    pub caching_strategy: CachingStrategy,
    /// The validation level, which defaults to 'full'
    pub validation_level: ValidationLevel,
    pub memory_cache_size: usize,
}

impl Relatable {
    /// Connect to a relatable database at the given path, or, if not given, at the location
    /// indicated by the environment variable RLTBL_CONNECTION, or, if that is not given,
    /// at [RLTBL_DEFAULT_DB]
    pub async fn connect(path: Option<&str>, caching_strategy: &CachingStrategy) -> Result<Self> {
        tracing::trace!("Relatable::connect({path:?}, {caching_strategy:?})");
        let root = std::env::var("RLTBL_ROOT").unwrap_or_default();
        // Set up database connection.
        let readonly = match std::env::var("RLTBL_READONLY") {
            Ok(value) if value.to_lowercase() != "false" => true,
            _ => false,
        };
        let path = match path {
            Some(path) => path.to_string(),
            None => {
                match std::env::var_os("RLTBL_CONNECTION").and_then(|p| Some(p.into_string())) {
                    Some(Ok(path)) => path,
                    _ => RLTBL_DEFAULT_DB.to_string(),
                }
            }
        };
        if !path.starts_with("postgresql://") {
            let file = FilePath::new(&path);
            if !file.exists() {
                return Err(RelatableError::InitError(
                    "First create a database with `rltbl init`".into(),
                )
                .into());
            }
        }
        let (connection, _) = DbConnection::connect(&path).await?;
        Ok(Self {
            root,
            readonly,
            connection,
            // minijinja: env,
            default_limit: DEFAULT_LIMIT,
            max_limit: MAX_LIMIT,
            caching_strategy: *caching_strategy,
            validation_level: ValidationLevel::Full,
            memory_cache_size: match caching_strategy {
                CachingStrategy::Memory(size) => {
                    let mut cache = CACHE.lock().expect("Could not lock cache");
                    let current_capacity = cache.capacity();
                    if current_capacity < *size {
                        cache.reserve(*size - current_capacity);
                    }
                    *size
                }
                _ => 0,
            },
        })
    }

    /// Initialize a [relatable](crate) database at the given path, or, if not given, at
    /// the location indicated by the environment variable RLTBL_CONNECTION, or, if that is not
    /// given, at [RLTBL_DEFAULT_DB]. Overwrites an existing database if `force` is set to true.
    pub async fn init(
        force: &bool,
        path: Option<&str>,
        caching_strategy: &CachingStrategy,
    ) -> Result<Self> {
        tracing::trace!("Relatable::init({force:?}, {path:?}, {caching_strategy:?})");
        let path = match path {
            Some(path) => path.to_string(),
            None => {
                match std::env::var_os("RLTBL_CONNECTION").and_then(|p| Some(p.into_string())) {
                    Some(Ok(path)) => path,
                    _ => RLTBL_DEFAULT_DB.to_string(),
                }
            }
        };
        if !path.starts_with("postgresql://") {
            let dir: &std::path::Path =
                FilePath::new(&path)
                    .parent()
                    .ok_or(RelatableError::InputError(
                        "Parent path must be defined".to_string(),
                    ))?;
            if !dir.exists() {
                std::fs::create_dir_all(&dir)?;
                tracing::info!("Created '{dir:?}' directory");
            }
            let file = FilePath::new(&path);
            if file.exists() {
                if *force {
                    std::fs::remove_file(&file)?;
                    tracing::info!("Removed '{file:?}' file");
                } else {
                    return Err(RelatableError::InitError(format!(
                        "File {file:?} already exists. Use --force to overwrite"
                    ))
                    .into());
                }
            }
            File::create(&path)?;
        }

        // Create the meta tables:
        let rltbl = Relatable::connect(Some(&path), caching_strategy).await?;
        let ddl = sql::generate_meta_tables_ddl(*force, &rltbl.connection.kind());
        for sql in ddl {
            rltbl.connection.query(&sql, None).await?;
        }

        Ok(rltbl)
    }

    /// Build a demonstration database. Based on <https://github.com/allisonhorst/palmerpenguins>.
    pub async fn build_demo(
        database: Option<&str>,
        force: &bool,
        size: usize,
        caching_strategy: &CachingStrategy,
    ) -> Result<Self> {
        tracing::trace!(
            "Relatable::build_demo({database:?}, {force}, {size}, {caching_strategy:?})"
        );
        let rltbl = Relatable::init(force, database.as_deref(), caching_strategy).await?;

        rltbl.create_demo_column_table(force).await?;
        rltbl.create_demo_datatype_table(force).await?;
        rltbl.create_penguin_table(None, force, size).await?;
        rltbl.create_island_table(None, force).await?;
        Ok(rltbl)
    }

    /// Create a demonstration table similar to the penguin table, but with the given name,
    /// and add `size` rows of data to it. Drop the table first if `force` is set.
    pub async fn create_penguin_table(
        &self,
        table: Option<&str>,
        force: &bool,
        size: usize,
    ) -> Result<()> {
        tracing::trace!("create_penguin_table({self:?}, {table:?}, {force}, {size})");
        let table = match table {
            Some(table) => table,
            None => "penguin",
        };
        if *force {
            if let DbKind::Postgres = self.connection.kind() {
                self.connection
                    .query(&format!(r#"DROP TABLE IF EXISTS "{table}" CASCADE"#), None)
                    .await?;
            }
        }

        let sql =
            format!(r#"INSERT INTO "table" ("table", "path") VALUES ('{table}', '{table}.tsv')"#);
        self.connection.query(&sql, None).await?;

        let pkey_clause = match self.connection.kind() {
            DbKind::Sqlite => "INTEGER PRIMARY KEY AUTOINCREMENT",
            DbKind::Postgres => "SERIAL PRIMARY KEY",
        };

        // Create the demo table:
        let sql = format!(
            r#"CREATE TABLE "{table}" (
             _id {pkey_clause},
             _order INTEGER UNIQUE,
             study_name TEXT,
             sample_number INTEGER,
             species TEXT,
             island TEXT,
             individual_id TEXT,
             bill_length REAL,
             bill_depth NUMERIC,
             body_mass BIGINT
           )"#,
        );
        self.connection.query(&sql, None).await?;

        let mut ddl = vec![];
        sql::add_metacolumn_trigger_ddl(&mut ddl, table, &self.connection.kind());
        if let CachingStrategy::Trigger = self.caching_strategy {
            sql::add_caching_trigger_ddl(&mut ddl, table, &self.connection.kind());
        }
        for sql in ddl {
            self.connection.query(&sql, None).await?;
        }
        // Populate the demo table with random data.
        let islands = vec!["Biscoe", "Dream", "Torgersen"];
        let mut rng = StdRng::seed_from_u64(0);
        let sql_first_part = format!(r#"INSERT INTO "{table}" VALUES "#);
        let mut sql_value_parts = vec![];
        let mut sql_param = SqlParam::new(&self.connection.kind());
        let mut param_values = vec![];
        let max_params = match self.connection.kind() {
            DbKind::Sqlite => sql::MAX_PARAMS_SQLITE,
            DbKind::Postgres => sql::MAX_PARAMS_POSTGRES,
        };
        for i in 0..size {
            if (param_values.len() + 8) >= max_params {
                let sql = format!(
                    "{sql_first_part} {sql_value_part}",
                    sql_value_part = sql_value_parts.join(", ")
                );
                let values_so_far = json!(param_values);
                self.connection.query(&sql, Some(&values_so_far)).await?;
                tracing::info!(
                    "{num_rows} rows loaded to table '{table}'",
                    num_rows = i - 1
                );
                param_values.clear();
                sql_value_parts.clear();
                sql_param.reset();
            }

            let id = i + 1;
            let order = id * NEW_ORDER_MULTIPLIER;
            let island = islands.iter().choose(&mut rng);
            let bill_length = rng.gen_range(300..500) as f64 / 10.0;
            let bill_depth = rng.gen_range(200..400) as f64 / 10.0;
            let body_mass = rng.gen_range(1000..5000);
            sql_value_parts.push(format!(
                "({sql_param_list_1}, 'FAKE123', {lone_sql_param}, 'Pygoscelis adeliae', \
                 {sql_param_list_2})",
                sql_param_list_1 = sql_param.get_as_list(2),
                lone_sql_param = sql_param.next(),
                sql_param_list_2 = sql_param.get_as_list(5),
            ));
            param_values.push(json!(id));
            param_values.push(json!(order));
            param_values.push(json!(id));
            param_values.push(json!(island));
            param_values.push(json!(format!("N{}A{}", (i / 2) + 1, (i % 2) + 1)));
            param_values.push(json!(bill_length));
            param_values.push(json!(bill_depth));
            param_values.push(json!(body_mass));
        }
        if param_values.len() > 0 {
            let sql = format!(
                "{sql_first_part} {sql_value_part}",
                sql_value_part = sql_value_parts.join(", ")
            );
            let param_values = json!(param_values);
            self.connection.query(&sql, Some(&param_values)).await?;
        }

        Ok(())
    }

    /// Create a demonstration table similar to the island table, but with the given name,
    /// and add `size` rows of data to it. Drop the table first if `force` is set.
    pub async fn create_island_table(&self, table: Option<&str>, force: &bool) -> Result<()> {
        tracing::trace!("create_island_table({self:?}, {table:?}, {force})");
        let table = match table {
            Some(table) => table,
            None => "island",
        };
        if *force {
            if let DbKind::Postgres = self.connection.kind() {
                self.connection
                    .query(&format!(r#"DROP TABLE IF EXISTS "{table}" CASCADE"#), None)
                    .await?;
            }
        }

        let sql =
            format!(r#"INSERT INTO "table" ("table", "path") VALUES ('{table}', '{table}.tsv')"#);
        self.connection.query(&sql, None).await?;

        let pkey_clause = match self.connection.kind() {
            DbKind::Sqlite => "INTEGER PRIMARY KEY AUTOINCREMENT",
            DbKind::Postgres => "SERIAL PRIMARY KEY",
        };

        // Create the demo table:
        let sql = format!(
            r#"CREATE TABLE "{table}" (
                 _id {pkey_clause},
                 _order INTEGER UNIQUE,
                 island_id INTEGER,
                 island TEXT
               )"#,
        );
        self.connection.query(&sql, None).await?;

        let mut ddl = vec![];
        sql::add_metacolumn_trigger_ddl(&mut ddl, table, &self.connection.kind());
        if let CachingStrategy::Trigger = self.caching_strategy {
            sql::add_caching_trigger_ddl(&mut ddl, table, &self.connection.kind());
        }
        for sql in ddl {
            self.connection.query(&sql, None).await?;
        }

        let sql = format!(
            r#"INSERT INTO "{table}" ("island_id", "island")
               VALUES (1, 'Torgersen'), (2, 'Biscoe'), (3, 'Dream')"#
        );

        self.connection.query(&sql, None).await?;
        Ok(())
    }

    /// Create the datatype table for the demonstration database
    pub async fn create_demo_datatype_table(&self, force: &bool) -> Result<()> {
        tracing::trace!("create_demo_datatype_table({self:?}, {force})");
        if *force {
            if let DbKind::Postgres = self.connection.kind() {
                self.connection
                    .query(r#"DROP TABLE IF EXISTS "datatype" CASCADE"#, None)
                    .await?;
            }
        }

        let pkey_clause = match self.connection.kind() {
            DbKind::Sqlite => "INTEGER PRIMARY KEY AUTOINCREMENT",
            DbKind::Postgres => "SERIAL PRIMARY KEY",
        };

        let sql = format!(
            r#"CREATE TABLE "datatype" (
             _id {pkey_clause},
             _order INTEGER UNIQUE,
             "datatype" TEXT,
             "description" TEXT,
             "parent" TEXT,
             "condition" TEXT,
             "sql_type" TEXT,
             "format" TEXT
           )"#,
        );
        self.connection.query(&sql, None).await?;

        let mut ddl = vec![];
        sql::add_metacolumn_trigger_ddl(&mut ddl, "datatype", &self.connection.kind());
        for sql in ddl {
            self.connection.query(&sql, None).await?;
        }

        let datatype_contents = [
            json!({
                "datatype": "decimal",
                "description": "A decimal number",
                "parent": "",
                "condition": "",
                "sql_type": "NUMERIC",
                "format": "%.1f"
            }),
            json!({
                "datatype": "study_name",
                "description": "",
                "parent": "text",
                "condition": "in(FAKE123, FAKE456)",
                "sql_type": "",
                "format": ""
            }),
        ]
        .iter()
        .map(|content| JsonRow {
            content: content.as_object().expect("Not a map").clone(),
        })
        .collect::<Vec<_>>();

        let mut sql_param_gen = SqlParam::new(&self.connection.kind());
        let mut param_values = vec![];
        let mut get_param = |row: &JsonRow, cname: &str| -> Result<String> {
            match row.get_value(cname)? {
                JsonValue::Null => Ok("NULL".to_string()),
                JsonValue::String(value) => {
                    param_values.push(value.to_string());
                    Ok(sql_param_gen.next().to_string())
                }
                _ => panic!("Invalid value type for datatype table"),
            }
        };
        let mut value_clauses = vec![];
        for row in &datatype_contents {
            let s1 = get_param(row, "datatype")?;
            let s2 = get_param(row, "description")?;
            let s3 = get_param(row, "parent")?;
            let s4 = get_param(row, "condition")?;
            let s5 = get_param(row, "sql_type")?;
            let s6 = get_param(row, "format")?;
            value_clauses.push(format!("({s1}, {s2}, {s3}, {s4}, {s5}, {s6})"));
        }

        let sql = format!(
            r#"INSERT INTO "datatype"
               ("datatype", "description", "parent", "condition", "sql_type", "format")
               VALUES {values}"#,
            values = value_clauses.join(", ")
        );
        let param_values = json!(param_values);
        self.connection.query(&sql, Some(&param_values)).await?;
        Ok(())
    }

    /// Create the column table for the demonstration database
    pub async fn create_demo_column_table(&self, force: &bool) -> Result<()> {
        tracing::trace!("create_demo_column_table({self:?}, {force})");
        if *force {
            if let DbKind::Postgres = self.connection.kind() {
                self.connection
                    .query(r#"DROP TABLE IF EXISTS "column" CASCADE"#, None)
                    .await?;
            }
        }

        let pkey_clause = match self.connection.kind() {
            DbKind::Sqlite => "INTEGER PRIMARY KEY AUTOINCREMENT",
            DbKind::Postgres => "SERIAL PRIMARY KEY",
        };

        let sql = format!(
            r#"CREATE TABLE "column" (
             _id {pkey_clause},
             _order INTEGER UNIQUE,
             "table" TEXT,
             "column" TEXT,
             "label" TEXT,
             "description" TEXT,
             "datatype" TEXT,
             "nulltype" TEXT,
             "structure" TEXT
           )"#,
        );
        self.connection.query(&sql, None).await?;

        let mut ddl = vec![];
        sql::add_metacolumn_trigger_ddl(&mut ddl, "column", &self.connection.kind());
        for sql in ddl {
            self.connection.query(&sql, None).await?;
        }

        let column_contents = [
            json!({
                "table": "penguin",
                "column": "study_name",
                "label": "study name",
                "datatype": "study_name",
            }),
            json!({
                "table": "penguin",
                "column": "sample_number",
                "label": "sample number",
                "description": "a sample number",
                "datatype": "integer",
            }),
            json!({
                "table": "penguin",
                "column": "species",
                "label": "species",
                "nulltype": "empty",
            }),
            json!({
                "table": "penguin",
                "column": "island",
                "label": "island",
                "datatype": "text",
                "structure": "from(island.island)",
            }),
            json!({
                "table": "penguin",
                "column": "individual_id",
                "label": "individual id",
                "nulltype": "empty",
                "datatype": "text",
            }),
            json!({
                "table": "penguin",
                "column": "bill_length",
                "label": "bill length (mm)",
                "datatype": "decimal",
            }),
            json!({
                "table": "penguin",
                "column": "bill_depth",
                "label": "bill depth (mm)",
                "datatype": "decimal",
            }),
            json!({
                "table": "penguin",
                "column": "body_mass",
                "label": "body mass (g)",
                "nulltype": "empty",
                "datatype": "integer",
            }),
        ]
        .iter()
        .map(|content| JsonRow {
            content: content.as_object().expect("Not a map").clone(),
        })
        .collect::<Vec<_>>();

        let mut sql_param_gen = SqlParam::new(&self.connection.kind());
        let mut param_values = vec![];
        let mut get_param = |row: &JsonRow, cname: &str| -> Result<String> {
            match row.get_value(cname).unwrap_or_default() {
                JsonValue::Null => Ok("NULL".to_string()),
                JsonValue::String(value) => {
                    param_values.push(value.to_string());
                    Ok(sql_param_gen.next().to_string())
                }
                _ => panic!("Invalid value type for column table"),
            }
        };
        let mut value_clauses = vec![];
        for row in &column_contents {
            let s1 = get_param(row, "table")?;
            let s2 = get_param(row, "column")?;
            let s3 = get_param(row, "label")?;
            let s4 = get_param(row, "description")?;
            let s5 = get_param(row, "nulltype")?;
            let s6 = get_param(row, "datatype")?;
            let s7 = get_param(row, "structure")?;
            value_clauses.push(format!("({s1}, {s2}, {s3}, {s4}, {s5}, {s6}, {s7})"));
        }

        let sql = format!(
            r#"INSERT INTO "column"
               ("table", "column", "label", "description", "nulltype", "datatype", "structure")
               VALUES {values}"#,
            values = value_clauses.join(", ")
        );
        let param_values = json!(param_values);
        self.connection.query(&sql, Some(&param_values)).await?;
        Ok(())
    }

    /// Create a tableset for the demonstration database
    pub async fn create_demo_tableset(&self, force: &bool, size: usize) -> Result<()> {
        tracing::trace!("create_demo_tableset({self:?}, {force}, {size})");
        if *force {
            if let DbKind::Postgres = self.connection.kind() {
                self.connection
                    .query(&format!(r#"DROP TABLE IF EXISTS "study" CASCADE"#), None)
                    .await?;
                self.connection
                    .query(&format!(r#"DROP TABLE IF EXISTS "penguin" CASCADE"#), None)
                    .await?;
                self.connection
                    .query(&format!(r#"DROP TABLE IF EXISTS "egg" CASCADE"#), None)
                    .await?;
            }
        }

        let sql = r#"INSERT INTO "table" ('table', 'path') VALUES ('tableset', 'tableset.tsv')"#;
        self.connection.query(sql, None).await.unwrap();

        // Create the tableset table.
        let sql = r#"CREATE TABLE tableset (
              _id INTEGER PRIMARY KEY AUTOINCREMENT,
              _order INTEGER UNIQUE,
              tableset TEXT,
              left_table TEXT,
              left_column TEXT,
              right_table TEXT,
              right_column TEXT
            )"#;
        self.connection.query(sql, None).await.unwrap();

        let sql = r#"INSERT INTO "tableset" VALUES
              (1, 1000, 'combined', NULL, NULL, 'study', 'study_name'),
              (2, 2000, 'combined', 'study', 'study_name', 'penguin', 'individual_id'),
              (3, 3000, 'combined', 'penguin', 'individual_id', 'egg', 'egg_id')
            "#;
        self.connection.query(sql, None).await.unwrap();

        let sql = r#"INSERT INTO "table" ('table', 'path') VALUES ('study', 'study.tsv')"#;
        self.connection.query(sql, None).await.unwrap();

        // Create the study table.
        let sql = r#"CREATE TABLE study (
              _id INTEGER PRIMARY KEY AUTOINCREMENT,
              _order INTEGER UNIQUE,
              study_name TEXT UNIQUE,
              description TEXT
            )"#;
        self.connection.query(sql, None).await.unwrap();

        let sql = r#"INSERT INTO study VALUES
            (0, 0, 'FAKE123', 'Fake Study 123')"#;
        self.connection.query(sql, None).await.unwrap();

        self.create_penguin_table(None, force, size).await?;

        let sql = r#"INSERT INTO "table" ('table', 'path') VALUES ('egg', 'egg.tsv')"#;
        self.connection.query(sql, None).await.unwrap();

        // Create the egg table.
        let sql = r#"CREATE TABLE egg (
      _id INTEGER PRIMARY KEY AUTOINCREMENT,
      _order INTEGER UNIQUE,
      egg_id TEXT UNIQUE,
      individual_id TEXT
    )"#;
        self.connection.query(sql, None).await.unwrap();

        let sql = r#"INSERT INTO egg VALUES
        (0, 0, 'E1', 'N1')"#;
        self.connection.query(sql, None).await.unwrap();

        Ok(())
    }

    // Drop all of the tables in the table table
    pub async fn drop_data_tables(&self) -> Result<()> {
        tracing::trace!("Relatable::drop_data_tables({self:?})");
        if !Table::table_exists("table", self).await? {
            tracing::warn!("Can't get list of tables to drop: The table table does not exist");
        } else {
            let mut tables = self.get_tables().await?;
            for (_, table) in tables.iter_mut() {
                let mut dependent_tables = table.get_dependent_tables(None, &self).await?;
                dependent_tables.reverse();
                for table in &mut dependent_tables {
                    table.drop_table(self).await?;
                }
                table.drop_table(self).await?;
            }
        }
        Ok(())
    }

    // Drop all of the meta tables
    pub async fn drop_meta_tables(&self) -> Result<()> {
        tracing::trace!("Relatable::drop_meta_tables({self:?})");
        for table_name in [
            "cache", "history", "change", "user", "message", "datatype", "column", "table",
        ] {
            let mut table = Table {
                name: table_name.to_string(),
                ..Default::default()
            };
            table.drop_table(self).await?;
        }
        Ok(())
    }

    // Drop all of the data tables and metatables in the database
    pub async fn drop_database(&self) -> Result<()> {
        tracing::trace!("Relatable::drop_database({self:?})");
        self.drop_data_tables().await?;
        self.drop_meta_tables().await?;
        Ok(())
    }

    /// Render this relatable instance in HTML according to the given template and context
    pub fn render<T: Serialize>(&self, template: &str, context: T) -> Result<String> {
        tracing::trace!("Relatable::render({template:?}, context)");
        // TODO: Optionally we should set up the environment once and store it,
        // but during development it's very convenient to rebuild every time.
        let mut env = Environment::new();

        // Load default template strings at compile time.
        let templates = IndexMap::from([
            ("page.html", include_str!("templates/page.html")),
            ("table.html", include_str!("templates/table.html")),
            ("row_menu.html", include_str!("templates/row_menu.html")),
            (
                "column_menu.html",
                include_str!("templates/column_menu.html"),
            ),
            ("cell_menu.html", include_str!("templates/cell_menu.html")),
        ]);

        // Load templates dynamically if src/templates/ exists,
        // otherwise use strings from compile time.
        // TODO: This should be a configuration option.
        let dir = std::env::var("RLTBL_TEMPLATES").unwrap_or("src/templates/".to_string());
        if FilePath::new(&dir).is_dir() {
            env.set_loader(path_loader(dir));
        };
        for (name, content) in templates {
            match env.get_template(name) {
                Ok(_) => (),
                Err(_) => env.add_template(name, content).unwrap(),
            }
        }

        env.get_template(template)?
            .render(context)
            .map_err(|e| e.into())
    }

    /// Use the given [Select] to fetch data from the database.
    pub async fn fetch(&self, select: &Select) -> Result<ResultSet> {
        tracing::trace!("Relatable::fetch({select:?})");

        // Get the table and columns information and use the given select to set the table's view:
        let mut table = Table::get_table(select.table_name.as_str(), self).await?;
        if select.view_name == format!("{}_default_view", table.name) || select.view_name == "" {
            table.set_view(self, "default").await?;
        } else if select.view_name == format!("{}_text_view", table.name) {
            table.set_view(self, "text").await?;
        } else {
            tracing::warn!(
                "Unsupported view name: '{}'. Falling back to default view",
                select.view_name
            );
            table.set_view(self, "default").await?;
        }
        let mut columns = table.columns.values().cloned().collect::<Vec<_>>();

        // Fetch the data
        let (statement, parameters) = select.to_sql(&self.connection.kind())?;
        let json_params = json!(parameters);
        let json_rows = self
            .connection
            .query(&statement, Some(&json_params))
            .await?;
        let count = json_rows.len();
        tracing::info!("Fetched {count} rows");

        // Filter out the table's columns that do not occur in the select:
        if select.select.len() > 0 {
            columns = columns
                .iter()
                .filter(|column| {
                    select.select.iter().any(|sel| match sel {
                        SelectField::Column {
                            table: select_table,
                            column: select_column,
                            ..
                        } => {
                            *select_column == column.name
                                && (select_table == "" || *select_table == table.name)
                        }
                        SelectField::Expression { alias, .. } => *alias == column.name,
                    })
                })
                .map(|c| c.clone())
                .collect();
        }

        // Return the data:
        let rows: Vec<Row> = json_rows.clone().vec_into();
        let total = self.count(&select).await?;
        Ok(ResultSet {
            select: select.clone(),
            statement,
            parameters,
            range: Range {
                count,
                total,
                start: (select.offset + 1) as u64,
                end: (select.offset + count) as u64,
            },
            table,
            columns,
            rows,
        })
    }

    /// Use the given [Select] to fetch data from the database.
    pub async fn fetch_rows(&self, select: &Select) -> Result<Vec<JsonRow>> {
        tracing::trace!("Relatable::fetch_rows({select:?})");
        let (statement, params) = select.to_sql(&self.connection.kind())?;
        let params = json!(params);
        self.connection.query(&statement, Some(&params)).await
    }

    /// Get the number of rows returned by this [Select] using the given caching strategy.
    pub async fn count(&self, select: &Select) -> Result<u64> {
        tracing::trace!("Relatable::count({select:?})");
        let (statement, params) = select.to_sql_count(&self.connection.kind())?;
        let params = json!(params);
        let json_rows = self
            .connection
            .cache(
                &statement,
                Some(&params),
                &select.get_tables().into_iter().collect(),
                &self.caching_strategy,
            )
            .await?;
        match json_rows.get(0) {
            Some(json_row) => json_row.get_unsigned("count"),
            None => Ok(0),
        }
    }

    /// Loads the given table from the given path. When `force` is set to true, deletes any
    /// existing table of the same name in the database first. When `validate` is set to true,
    /// Validates each row before loading it. Note that this function may panic.
    pub async fn load_table(&self, table_name: &str, path: &str, force: bool) {
        tracing::trace!("Relatable::load_table({table_name:?}, {path:?}, {force})");
        // Read the records from the given TSV file:
        let mut rdr = ReaderBuilder::new()
            .has_headers(false)
            .delimiter(b'\t')
            .from_reader(File::open(path).expect(&format!("Unable to open '{path}'")));
        let mut records = rdr.records();

        // Extract the headers from the first line of the file, which we will need for the CREATE
        // TABLE statement:
        let headers = {
            let headers = match records.next() {
                None => panic!("'{path}' is empty"),
                Some(record) => match record {
                    Err(err) => panic!("Error reading from '{path}': {err}"),
                    Ok(headers) => headers.iter().map(|s| s.to_string()).collect::<Vec<_>>(),
                },
            };
            for header in &headers {
                if header.trim().is_empty() {
                    panic!("One or more of the header fields is empty for table '{table_name}'");
                }
            }
            headers
        };

        let db_kind = self.connection.kind();

        // Add an entry corresponding to the table being loaded to the table table:
        if force {
            // Delete any messages associated with the table and then delete the table:
            self.delete_message(table_name, None, None, None, None)
                .await
                .expect("Error deleting messages");

            let sql = format!(
                r#"DELETE FROM "table" WHERE "table" = {sql_param}"#,
                sql_param = SqlParam::new(&db_kind).next(),
            );
            let params = json!([table_name]);
            self.connection
                .query(&sql, Some(&params))
                .await
                .expect("Error deleting from table table");
        }
        let sql = format!(
            r#"INSERT INTO "table" ("table", "path") VALUES ({sql_params})"#,
            sql_params = SqlParam::new(&db_kind).get_as_list(2)
        );
        let params = json!([table_name, path]);
        self.connection
            .query(&sql, Some(&params))
            .await
            .expect("Error inserting to table table");
        tracing::debug!("Table {table_name} (path: {path}) added to table table");

        // Initialize a new table struct and collect its columns configuration:
        let table = {
            let mut table = Table {
                name: table_name.to_string(),
                ..Default::default()
            };
            let table_columns = Table::get_column_table_columns(table_name, self)
                .await
                .expect(&format!("Error getting columns for table '{table_name}'"));
            for column_name in headers.iter() {
                let datatype = match table_columns.get(column_name) {
                    None => Datatype {
                        name: "text".to_string(),
                        ..Default::default()
                    },
                    Some(col) => col.datatype.clone(),
                };
                let column = Column {
                    name: column_name.to_string(),
                    table: table_name.to_string(),
                    datatype_hierarchy: datatype.get_all_ancestors(self).await.expect(&format!(
                        "Error getting datatype hierarchy for '{}'",
                        datatype.name
                    )),
                    datatype: datatype,
                    nulltype: table_columns
                        .get(column_name)
                        .and_then(|col| col.nulltype.clone()),
                    structure: table_columns
                        .get(column_name)
                        .and_then(|col| col.structure.clone()),
                    ..Default::default()
                };
                table.columns.insert(column_name.to_string(), column);
            }
            table
        };

        // Generate the SQL statements needed to create the table and execute them:
        for sql in sql::generate_table_ddl(&table, force, &db_kind, &self.caching_strategy)
            .expect("Error getting DDL")
        {
            self.connection
                .query(&sql, None)
                .await
                .expect("Error creating table");
        }

        // Insert the data into the table:
        let mut columns = vec!["_id".to_string(), "_order".to_string()];
        columns.append(
            &mut headers
                .iter()
                .map(|k| format!(r#"{k}"#))
                .collect::<Vec<_>>(),
        );
        let columns_line = columns
            .iter()
            .map(|k| format!(r#""{k}""#))
            .collect::<Vec<_>>()
            .join(", ");
        let mut id: u64 = 1;
        let mut order = id * NEW_ORDER_MULTIPLIER as u64;
        let sql_first_part = format!(r#"INSERT INTO "{table_name}" ({columns_line}) VALUES "#);
        let mut sql_value_parts = vec![];
        let mut sql_param_gen = SqlParam::new(&self.connection.kind());
        let mut param_values = vec![];
        let max_params = match db_kind {
            DbKind::Sqlite => sql::MAX_PARAMS_SQLITE,
            DbKind::Postgres => sql::MAX_PARAMS_POSTGRES,
        };
        while let Some(row) = records.next() {
            let row = row.expect("Error processing row");
            // We add 2 here because of _id and _order:
            if (param_values.len() + row.len() + 2) >= max_params {
                let sql = format!(
                    "{sql_first_part} {sql_value_part}",
                    sql_value_part = sql_value_parts.join(", ")
                );
                let values_so_far = json!(param_values);
                self.connection
                    .query(&sql, Some(&values_so_far))
                    .await
                    .expect("Error inserting to table");
                tracing::info!(
                    "{num_rows} rows loaded to table {table_name}",
                    num_rows = id - 1
                );
                param_values.clear();
                sql_value_parts.clear();
                sql_param_gen.reset()
            }

            let mut sql_params = vec![];
            param_values.push(json!(id));
            sql_params.push(sql_param_gen.next());
            param_values.push(json!(order));
            sql_params.push(sql_param_gen.next());
            let sql_params = {
                for (i, value) in row.iter().enumerate() {
                    let (column, nulltype) = {
                        // We add 2 here because of _id and _order:
                        let column = match columns.get(i + 2) {
                            Some(column) => column,
                            None => panic!("Unable to retrieve column {}", i + 2),
                        };
                        let nulltype = table
                            .columns
                            .get(column)
                            .expect(&format!("Column '{column}' not found"))
                            .nulltype
                            .to_owned();
                        (column, nulltype)
                    };
                    match nulltype {
                        Some(nulltype) if nulltype.name == "empty" && value == "" => {
                            sql_params.push("NULL".to_string());
                        }
                        _ => {
                            if let Some(nulltype) = nulltype {
                                if nulltype.name != "empty" {
                                    tracing::warn!("Nulltype '{}' not supported", nulltype.name);
                                }
                            }
                            // Use the value to create a cell:
                            let mut cell = {
                                let value = match serde_json::from_str::<JsonValue>(value) {
                                    Ok(JsonValue::Number(num)) => JsonValue::Number(num),
                                    _ => json!(value),
                                };
                                let value = JsonRow::nullify_value(&table, column, &value);
                                Cell {
                                    text: sql::json_to_string(&value),
                                    value: value,
                                    ..Default::default()
                                }
                            };

                            // Validate the cell and add any messages to the message table:
                            if self.validation_level != ValidationLevel::None {
                                cell.validate_sql_type(&table.get_config_for_column(column))
                                    .expect("Error validating cell");
                                for message in cell.messages.iter() {
                                    let (msg_id, msg) = self
                                        .add_message(
                                            "rltbl",
                                            &table.name,
                                            id,
                                            column,
                                            &cell.value,
                                            &message.level,
                                            &message.rule,
                                            &message.message,
                                        )
                                        .await
                                        .expect("Error adding message");
                                    tracing::debug!("Added message (ID {msg_id}): {msg:?}");
                                }
                            }

                            // Add the parameter for the value to the SQL insert statement:
                            if cell.has_sql_type_error() || cell.value == JsonValue::Null {
                                sql_params.push("NULL".to_string());
                            } else {
                                sql_params.push(sql_param_gen.next());
                                param_values.push(cell.value);
                            }
                        }
                    };
                }
                sql_params.join(", ")
            };
            // Add two extra SQL_PARAM for _id and _order:
            sql_value_parts.push(format!("({sql_params})"));
            id += 1;
            order += NEW_ORDER_MULTIPLIER as u64;
        }
        if param_values.len() > 0 {
            let sql = format!(
                "{sql_first_part} {sql_value_part}",
                sql_value_part = sql_value_parts.join(", ")
            );
            let param_values = json!(param_values);
            self.connection
                .query(&sql, Some(&param_values))
                .await
                .expect(&format!("Error inserting to {table_name}"));
            tracing::info!(
                "{num_rows} rows loaded to table {table_name}",
                num_rows = id - 1
            );
        }

        if self.validation_level == ValidationLevel::Full {
            self.validate_table(&table)
                .await
                .expect("Error validating table");
            let dependent_tables = table
                .get_dependent_tables(None, &self)
                .await
                .expect("Error getting dependent tables");
            for table in &dependent_tables {
                tracing::debug!("Validating dependent table '{}'", table.name);
                self.validate_structure_for_table(&table)
                    .await
                    .expect("Error validating table");
            }
        }

        self.commit_to_git().await.expect("Error committing to git");
    }

    /// Save all of the tables that have entries in the table table to the path indicated for each
    /// table there, unless `save_dir` has been given, in which case save them all there instead.
    pub async fn save_all(&self, save_dir: Option<&str>) -> Result<()> {
        tracing::trace!("Relatable::save_all({save_dir:?})");
        let sql = format!(
            r#"SELECT "table", "path" FROM "table" WHERE "path" {is_not} NULL"#,
            is_not = sql::is_not_clause(&self.connection.kind())
        );
        let table_rows = self.connection.query(&sql, None).await?;
        for table_row in table_rows {
            let table_name = table_row.get_string("table")?;
            let mut table = Table::get_table(&table_name, self).await?;
            table.set_view(self, "text").await?;

            let path = match save_dir {
                Some(save_dir) => format!("{save_dir}/{table_name}.tsv"),
                None => table_row.get_string("path")?,
            };
            let mut writer = WriterBuilder::new()
                .delimiter(b'\t')
                .quote_style(QuoteStyle::Never)
                .from_path(path)?;
            let header_row = self
                .fetch_columns(&table_name)
                .await?
                .iter()
                .map(|c| c.name.to_string())
                .collect::<Vec<_>>();
            writer.write_record(header_row.clone())?;

            let sql = format!(
                r#"SELECT {columns} FROM "{table_name}_text_view" ORDER BY "_order""#,
                columns = header_row
                    .iter()
                    .map(|c| format!(r#""{c}""#))
                    .collect::<Vec<_>>()
                    .join(", ")
            );
            let data_rows = self.connection.query(&sql, None).await?;
            for data_row in data_rows {
                let values = {
                    let mut str_values = vec![];
                    for (column, value) in data_row.content.iter() {
                        match value {
                            JsonValue::String(s) => str_values.push(s.to_string()),
                            JsonValue::Number(n) => str_values.push(n.to_string()),
                            JsonValue::Null => {
                                match &table
                                    .columns
                                    .get(column)
                                    .ok_or(RelatableError::InputError(format!(
                                        "Column '{column}' not found"
                                    )))?
                                    .nulltype
                                {
                                    // Note that the behaviour for the 'empty' nulltype happens
                                    // to be the same as that for no nulltype, but in general
                                    // that won't be true for every nulltype.
                                    Some(nulltype) if nulltype.name == "empty" => {
                                        str_values.push("".to_string());
                                    }
                                    Some(unsup) => {
                                        tracing::warn!("Unsupported nulltype: '{}'", unsup.name);
                                        str_values.push("".to_string());
                                    }
                                    None => {
                                        str_values.push("".to_string());
                                    }
                                };
                            }
                            _ => {
                                return Err(RelatableError::DataError(format!(
                                    "Value {value} is not a string, number or NULL"
                                ))
                                .into());
                            }
                        }
                    }
                    str_values
                };
                writer.write_record(values)?;
            }
        }

        Ok(())
    }

    /// Save all of the tables and commit the changes to git.
    pub async fn commit_to_git(&self) -> Result<()> {
        tracing::trace!("Relatable::commit_to_git()");
        let author = match std::env::var("RLTBL_GIT_AUTHOR") {
            Err(err) => match err {
                std::env::VarError::NotPresent => {
                    tracing::debug!("Not committing to git because RLTBL_GIT_AUTHOR not defined");
                    return Ok(());
                }
                _ => {
                    return Err(RelatableError::InputError(format!(
                        "Could not read from the environment: {err}"
                    ))
                    .into())
                }
            },
            Ok(author) => author,
        };
        tracing::info!("Committing to git on behalf of RLTBL_GIT_AUTHOR: '{author}'");

        // Save all the tables:
        self.save_all(None).await?;

        // Get the git status:
        let status = git::get_status()?;
        if status.behind != 0 {
            return Err(RelatableError::GitError(
                "Refusing to commit to a local repository that is behind the remote".to_string(),
            )
            .into());
        }

        // Possibly only amend the last commit, if it is by the same author and performed
        // on the same day:
        let (last_commit_author, days_ago) = git::get_last_commit_info()?;
        let is_amendment = (last_commit_author == author) && (days_ago < 1);

        // Stage any modified table files that have a path in the table table:
        let sql = format!(
            r#"SELECT "path" FROM "table" WHERE "path" {is_not} NULL"#,
            is_not = sql::is_not_clause(&self.connection.kind()),
        );
        let paths = self
            .connection
            .query(&sql, None)
            .await?
            .iter()
            .map(|row| row.get_string("path").expect("No 'path' found"))
            .collect::<Vec<_>>();
        git::add(&paths)?;

        // Finally, commit to git:
        git::commit("commit by rltbl", &author, is_amendment)?;
        Ok(())
    }

    /// Get the details of the last change made by the user from the change table.
    fn _get_last_change_for_user(
        &self,
        tx: &mut DbTransaction<'_>,
        user: &str,
        action: &ChangeAction,
    ) -> Result<Option<(u64, ChangeSet)>> {
        tracing::trace!("Relatable::_get_last_change_for_user(tx, {user:?}, {action:?})");
        let mut sql_param = SqlParam::new(&tx.kind());
        let sql = format!(
            r#"SELECT "change_id", "user", "table", "description", "content"
               FROM "change"
               WHERE "user" = {sql_param_1} AND "action" = {sql_param_2}
               ORDER BY "change_id" DESC LIMIT 1"#,
            sql_param_1 = sql_param.next(),
            sql_param_2 = sql_param.next(),
        );
        let params = json!([user, format!("{action}")]);
        let records = tx.query(&sql, Some(&params))?;
        match records.len() {
            0 => Ok(None),
            _ => {
                let change_id = records[0].get_unsigned("change_id")?;
                let user = records[0].get_string("user")?;
                let table = records[0].get_string("table")?;
                let description = records[0].get_string("description")?;
                let content = records[0].get_string("content")?;
                let changes = Change::many_from_str(&content)?;
                Ok(Some((
                    change_id,
                    ChangeSet {
                        action: *action,
                        table: table,
                        user: user,
                        description: description,
                        changes: changes,
                    },
                )))
            }
        }
    }

    /// Record the given [ChangeSet] to the change and history tables.
    pub fn record_changeset(
        &self,
        changeset: &ChangeSet,
        tx: &mut DbTransaction<'_>,
    ) -> Result<()> {
        tracing::trace!("Relatable::record_changeset({changeset:?}, tx)");
        let user = changeset.user.clone();
        let action = changeset.action.to_string();
        let table = changeset.table.clone();
        let description = changeset.description.clone();

        // Begin by getting the current last change_id for this user, which we may need to look
        // up previous values of the row's columns in the history table later:
        let old_change_id = match &changeset.action {
            ChangeAction::Undo => {
                let (change_id, _) = self
                    ._get_last_change_for_user(tx, &changeset.user, &ChangeAction::Do)?
                    .ok_or(RelatableError::DataError(
                        "No action for user found".to_string(),
                    ))?;
                Some(change_id)
            }
            ChangeAction::Redo => {
                let (change_id, _) = self
                    ._get_last_change_for_user(tx, &changeset.user, &ChangeAction::Undo)?
                    .ok_or(RelatableError::DataError(
                        "No undo for user found".to_string(),
                    ))?;
                Some(change_id)
            }
            ChangeAction::Do => None,
        };

        // Now write the current change, which will generate a new last change_id:
        let statement = format!(
            r#"INSERT INTO change("user", "action", "table", "description", "content")
               VALUES ({sql_params})
               RETURNING change_id"#,
            sql_params = SqlParam::new(&tx.kind()).get_as_list(5)
        );
        let content = to_value(&changeset.changes).unwrap_or_default();
        let params = json!([user, action, table, description, content]);
        let change_id = tx.query_value(&statement, Some(&params))?;
        let change_id = change_id
            .ok_or(RelatableError::DataError(
                "Expected a change_id".to_string(),
            ))?
            .as_u64()
            .ok_or(RelatableError::DataError("Expected an integer".to_string()))?;

        for change in &changeset.changes {
            match change {
                Change::Update {
                    row,
                    column,
                    before,
                    after,
                } => {
                    let sql = format!(
                        r#"INSERT INTO "history"
                           ("change_id", "table", "row", "before", "after")
                           VALUES ({sql_params})
                           RETURNING "history_id""#,
                        sql_params = SqlParam::new(&tx.kind()).get_as_list(5)
                    );
                    let before = json!({column: before}).to_string();
                    let after = json!({column: after}).to_string();
                    let params = json!([change_id, table, row, before, after]);
                    tx.query_value(&sql, Some(&params))?;
                }
                Change::Add { row, after: _ } => {
                    // If the row has just been newly added, it will be found in the table,
                    // otherwise we will use the old_change_id to look for it in the history
                    // table:
                    let json_row = match Table::_get_row(&table, *row, tx)? {
                        Some(json_row) => json_row,
                        None => match old_change_id {
                            Some(change_id) => {
                                let sql = format!(
                                    r#"SELECT "before"
                                         FROM "history"
                                        WHERE "change_id" = {sql_param}"#,
                                    sql_param = SqlParam::new(&tx.kind()).next()
                                );
                                let params = json!([change_id]);
                                let before = tx
                                    .query_one(&sql, Some(&params))?
                                    .ok_or(RelatableError::DataError(format!(
                                        "No history row found with change_id {change_id}"
                                    )))?
                                    .get_string("before")?;
                                let before = match serde_json::from_str::<JsonValue>(&before) {
                                    Err(err) => return Err(err.into()),
                                    Ok(JsonValue::Object(o)) => o,
                                    Ok(_) => {
                                        return Err(RelatableError::InputError(
                                            "The content parameter is not an object".to_string(),
                                        )
                                        .into());
                                    }
                                };
                                JsonRow { content: before }
                            }
                            None => {
                                return Err(RelatableError::DataError(format!(
                                    "Row {row} not found"
                                ))
                                .into())
                            }
                        },
                    };
                    let sql = format!(
                        r#"INSERT INTO "history"
                           ("change_id", "table", "row", "after")
                           VALUES ({sql_params})
                           RETURNING "history_id""#,
                        sql_params = SqlParam::new(&tx.kind()).get_as_list(4)
                    );
                    let json_row_str = json!(json_row.content).to_string();
                    let params = json!([change_id, table, row, json_row_str]);
                    tx.query_value(&sql, Some(&params))?;
                }
                Change::Move {
                    row,
                    from_after: _,
                    to_after: _,
                } => {
                    let sql = format!(
                        r#"INSERT INTO "history"
                           ("change_id", "table", "row")
                           VALUES ({sql_params})
                           RETURNING "history_id""#,
                        sql_params = SqlParam::new(&tx.kind()).get_as_list(3)
                    );
                    let params = json!([change_id, table, row]);
                    tx.query_value(&sql, Some(&params))?;
                }
                Change::Delete { row, after: _ } => {
                    let json_row = match Table::_get_row(&table, *row, tx)? {
                        Some(json_row) => json_row,
                        None => {
                            // It must be there since we supposedly just added it, so if it is
                            // not found return an error.
                            return Err(
                                RelatableError::DataError(format!("Row {row} not found")).into()
                            );
                        }
                    };
                    let sql = format!(
                        r#"INSERT INTO "history"
                           ("change_id", "table", "row", "before")
                           VALUES ({sql_params})
                           RETURNING "history_id""#,
                        sql_params = SqlParam::new(&tx.kind()).get_as_list(4)
                    );
                    let json_row_str = json!(json_row.content).to_string();
                    let params = json!([change_id, table, row, json_row_str]);
                    tx.query_value(&sql, Some(&params))?;
                }
            };
        }

        // Possibly delete dirty entries from the cache in accordance with our caching strategy:
        match self.caching_strategy {
            // Trigger has the same behaviour as None here, since the database will be triggering
            // this step automatically every time the table is edited in that case.
            CachingStrategy::None | CachingStrategy::Trigger => (),
            CachingStrategy::Memory(_) => self.clear_mem_cache(&table),
            CachingStrategy::TruncateAll => Relatable::clear_cache(tx, None)?,
            CachingStrategy::Truncate => Relatable::clear_cache(tx, Some(&table))?,
        };

        Ok(())
    }

    /// Get information about the given user from the database and return it as an [Account]. If
    /// there is no user with the given username, return a default Account.
    pub async fn get_user(&self, username: &str) -> Account {
        tracing::trace!("Relatable::get_user({username:?})");
        let statement = format!(
            r#"SELECT "name", "color", "cursor", "datetime"
               FROM "user" WHERE name = '{username}' LIMIT 1"#
        );
        let user = self.connection.query_one(&statement, None).await;
        match user {
            Ok(user) => match user {
                Some(user) => Account {
                    name: username.to_string(),
                    color: user.get_string("color").expect("No 'color' found"),
                },
                None => Account {
                    ..Default::default()
                },
            },
            Err(err) => {
                tracing::warn!("Error while querying user table: '{err}'");
                Account {
                    ..Default::default()
                }
            }
        }
    }

    /// Returns a map with information about all of the users who have corresponding records in
    /// the user table.
    pub async fn get_users(&self) -> Result<IndexMap<String, UserCursor>> {
        tracing::trace!("Relatable::get_users()");
        let mut users = IndexMap::new();
        // let statement = format!(
        //     r#"SELECT "name", color", "cursor", "datetime" FROM "user" WHERE cursor IS NOT NULL
        //        AND "datetime" >= DATETIME('now', '-10 minutes')"#
        // );
        let statement = format!(
            r#"SELECT "name", "color", "cursor", "datetime"
               FROM "user"
               WHERE cursor {is_not} NULL"#,
            is_not = sql::is_not_clause(&self.connection.kind()),
        );
        let rows = self.connection.query(&statement, None).await?;
        for row in rows {
            let name = row.get_string("name")?;
            if name.trim() == "" {
                continue;
            }
            users.insert(
                name.clone(),
                UserCursor {
                    name: name.clone(),
                    color: row.get_string("color")?,
                    cursor: serde_json::from_str(&row.get_string("cursor")?)?,
                    datetime: row.get_string("datetime")?,
                },
            );
        }
        Ok(users)
    }

    /// Returns a list of the given table's columns, not including metacolumns
    pub async fn fetch_columns(&self, table_name: &str) -> Result<Vec<Column>> {
        tracing::trace!("Relatable::fetch_columns({table_name:?})");
        let table = Table::get_table(table_name, self).await?;
        Ok(table.columns.values().cloned().collect::<Vec<_>>())
    }

    /// Returns a list of the given table's columns, including metacolumns
    pub async fn fetch_all_columns(&self, table_name: &str) -> Result<Vec<Column>> {
        tracing::trace!("Relatable::fetch_all_columns({table_name:?})");
        let mut conn = self.connection.reconnect()?;
        // Begin a transaction:
        let mut tx = self.connection.begin(&mut conn).await?;

        let columns = {
            let (mut normal_columns, meta_columns) =
                Table::_collect_column_info(table_name, &mut tx)?;
            let mut all_columns = meta_columns;
            all_columns.append(&mut normal_columns);
            all_columns
        };

        // Commit the transaction:
        tx.commit()?;

        Ok(columns)
    }

    /// Returns a vector of the names of the tables that have entries in the table table
    pub async fn list_tables(&self) -> Result<Vec<String>> {
        tracing::trace!("Relatable::list_tables({self:?})");
        let statement = format!(r#"SELECT "table" FROM "table" ORDER BY _order"#);
        let rows = self.connection.query(&statement, None).await?;
        rows.iter().map(|row| row.get_string("table")).collect()
    }

    /// Returns all of the tables that have entries in the table table as a map from table names
    /// to Table structs.
    pub async fn get_tables(&self) -> Result<IndexMap<String, Table>> {
        tracing::trace!("Relatable::get_tables({self:?})");
        let mut tables = IndexMap::new();
        let statement = format!(
            r#"SELECT "_id", "_order", "table", "path",
                 (SELECT MAX(change_id)
                  FROM "history"
                  WHERE "history"."table" = "table"."table"
                 ) AS "_change_id"
               FROM "table""#
        );

        let rows = self.connection.query(&statement, None).await?;
        for row in rows {
            let name = row.get_string("table")?;
            tables.insert(
                name.clone(),
                Table {
                    name: name.clone(),
                    change_id: row
                        .content
                        .get("_change_id")
                        .and_then(|i| i.as_u64())
                        .unwrap_or_default() as u64,
                    columns: self
                        .fetch_columns(&name)
                        .await?
                        .into_iter()
                        .map(|column| (name.clone(), column))
                        .collect::<IndexMap<_, _>>(),
                    ..Default::default()
                },
            );
        }
        Ok(tables)
    }

    /// Returns a [Site] corresponding to the given username.
    pub async fn get_site(&self, username: &str) -> Site {
        tracing::trace!("Relatable::get_site({username:?})");
        let mut users = self.get_users().await.unwrap_or_default();
        users.shift_remove(username);
        Site {
            title: "RLTBL".to_string(),
            root: self.root.clone(),
            editable: !self.readonly,
            user: self.get_user(username).await,
            users,
            tables: self.list_tables().await.unwrap_or_default(),
        }
    }

    /// Updates the cursor field in the user table for the user associated with the given
    /// changeset.
    pub fn prepare_user_cursor(
        &self,
        changeset: &ChangeSet,
        tx: &mut DbTransaction<'_>,
    ) -> Result<()> {
        tracing::trace!("Relatable::prepare_user_cursor({changeset:?}, tx)");
        // Make sure the user is present in the user table
        let user = changeset.user.clone();
        let color = random_color::RandomColor::new().to_hex();
        let statement = format!(
            r#"SELECT 1 FROM "user" WHERE "name" = {sql_param}"#,
            sql_param = SqlParam::new(&tx.kind()).next()
        );
        let params = json!([user]);
        if let None = tx.query_value(&statement, Some(&params))? {
            let statement = format!(
                r#"INSERT INTO "user" ("name", "color") VALUES ({sql_params})"#,
                sql_params = SqlParam::new(&tx.kind()).get_as_list(2)
            );
            let params = json!([user, color]);
            tx.query(&statement, Some(&params))?;
        }

        // Update the user's cursor position.
        let mut cursor = changeset.to_cursor()?;
        match changeset.action {
            ChangeAction::Undo | ChangeAction::Redo => match changeset.changes.first() {
                Some(Change::Delete { row, after: _ }) => {
                    cursor.row = Table::_get_previous_row_id(&changeset.table, *row, tx)?;
                }
                _ => (),
            },
            ChangeAction::Do => (),
        };

        let mut sql_param = SqlParam::new(&tx.kind());
        let statement = format!(
            r#"UPDATE "user"
               SET "cursor" = {sql_param_1}, "datetime" = CURRENT_TIMESTAMP
               WHERE "name" = {sql_param_2}"#,
            sql_param_1 = sql_param.next(),
            sql_param_2 = sql_param.next(),
        );
        let params = json!([to_value(cursor).unwrap_or_default(), user]);
        tx.query_value(&statement, Some(&params))?;

        Ok(())
    }

    /// Get the last set of changes that can be redone for the given user
    pub async fn get_last_redoable_changeset_for_user(
        &self,
        user: &str,
    ) -> Result<Option<(u64, ChangeSet)>> {
        tracing::trace!("Relatable::get_last_redoable_changeset_for_user({user:?})");
        let history = self.get_user_history(user, Some(1)).await?;
        match history.changes_undone_stack.first() {
            None => Ok(None),
            Some(change) => {
                let change_id = change.get_unsigned("change_id")?;
                let content = change.get_string("content")?;
                let changes = Change::many_from_str(&content)?;
                Ok(Some((
                    change_id,
                    ChangeSet {
                        action: ChangeAction::from_str(&change.get_string("action")?)?,
                        table: change.get_string("table")?,
                        user: change.get_string("user")?,
                        description: change.get_string("user")?,
                        changes: changes,
                    },
                )))
            }
        }
    }

    /// Get the last set of changes that can be undone for the given user
    pub async fn get_last_undoable_changeset_for_user(
        &self,
        user: &str,
    ) -> Result<Option<(u64, ChangeSet)>> {
        tracing::trace!("Relatable::get_last_undoable_changeset_for_user({user:?})");
        let history = self.get_user_history(user, Some(1)).await?;
        match history.changes_done_stack.first() {
            None => Ok(None),
            Some(change) => {
                let change_id = change.get_unsigned("change_id")?;
                let content = change.get_string("content")?;
                let changes = Change::many_from_str(&content)?;
                Ok(Some((
                    change_id,
                    ChangeSet {
                        action: ChangeAction::from_str(&change.get_string("action")?)?,
                        table: change.get_string("table")?,
                        user: change.get_string("user")?,
                        description: change.get_string("user")?,
                        changes: changes,
                    },
                )))
            }
        }
    }

    /// Return a [History] for the given user with at most `context` (or [HISTORY_MAX] if this
    /// is not given) undoable and/or redoable previous changes.
    pub async fn get_user_history(&self, user: &str, context: Option<usize>) -> Result<History> {
        tracing::trace!("Relatable::get_user_history({user:?}, {context:?})");
        fn content_to_json_row(content: &str) -> Result<JsonRow> {
            tracing::debug!("Entering content_to_json_row(content: {content})");
            match serde_json::from_str::<JsonValue>(content) {
                Ok(content) => match content
                    .as_array()
                    .and_then(|a| a.first())
                    .and_then(|o| o.as_object())
                {
                    Some(object) => Ok(JsonRow {
                        content: object.clone(),
                    }),
                    None => {
                        return Err(RelatableError::InputError(format!(
                            "Received invalid or empty content: {content}. Expected a non-empty \
                             object array."
                        ))
                        .into())
                    }
                },
                Err(err) => {
                    return Err(
                        RelatableError::InputError(format!("Error reading content: {err}")).into(),
                    )
                }
            }
        }

        fn on_the_same_target(change1: &JsonRow, change2: &JsonRow) -> Result<bool> {
            tracing::trace!("Relatable::on_the_same_target({change1:?}, {change2:?})");
            let change1 = content_to_json_row(&change1.get_string("content")?)?;
            let change2 = content_to_json_row(&change2.get_string("content")?)?;
            let row1 = change1.get_unsigned("row")?;
            let row2 = change2.get_unsigned("row")?;
            if row1 != row2 {
                return Ok(false);
            }
            if let Ok(column1) = change1.get_string("column") {
                if let Ok(column2) = change2.get_string("column") {
                    return Ok(column1 == column2);
                }
            }
            Ok(true)
        }

        fn prune_stacks(
            changes_done_stack: &Vec<JsonRow>,
            changes_undone_stack: &Vec<JsonRow>,
        ) -> (Vec<JsonRow>, Vec<JsonRow>) {
            tracing::trace!(
                "Relatable::prune_stacks({changes_done_stack:?}, {changes_undone_stack:?})"
            );
            let mut pruned_dones = vec![];
            let mut pruned_undones = vec![];
            for change in changes_done_stack.iter() {
                if !pruned_dones.iter().any(|done: &JsonRow| {
                    on_the_same_target(&done, &change).expect("Error looking for a common target")
                }) {
                    pruned_dones.push(change.clone());
                }
            }
            for change in changes_undone_stack.iter() {
                if !pruned_undones.iter().any(|undone: &JsonRow| {
                    on_the_same_target(&undone, &change).expect("Error looking for a common target")
                }) {
                    pruned_undones.push(change.clone());
                }
            }
            (pruned_dones, pruned_undones)
        }

        // TODO: Think about paging when there are a lot of change records to go through.
        let sql = format!(
            r#"SELECT "change_id", "user", "table", "description", "action", "content"
                 FROM "change"
                WHERE "user" = {sql_param}
                ORDER BY "change_id" DESC"#,
            sql_param = SqlParam::new(&self.connection.kind()).next()
        );
        let params = json!([user]);
        let history = self.connection.query(&sql, Some(&params)).await?;

        // Initialize the stacks to be returned and counters:
        let mut changes_done_stack = vec![];
        let mut changes_undone_stack = vec![];
        let mut do_redo_count: usize;
        let mut undo_count: usize;

        // Begin with the last change that was made:
        let (final_change, final_action) = match history.first() {
            None => return Ok(History::default()),
            Some(final_change) => {
                let final_action = ChangeAction::from_str(&final_change.get_string("action")?)?;
                (final_change, final_action)
            }
        };
        match final_action {
            ChangeAction::Do | ChangeAction::Redo => {
                do_redo_count = 1;
                undo_count = 0;
            }
            ChangeAction::Undo => {
                do_redo_count = 0;
                undo_count = 1;
            }
        };
        let mut change_to_push = final_change;
        let mut action_to_push = final_action;
        tracing::debug!("Setting the change to push ({action_to_push}) to: {change_to_push:?}");
        tracing::debug!(
            "The do/redo count is now {do_redo_count}, and the undo count is \
             {undo_count}."
        );

        // For each action, find the point where it began, and then place it onto changes_done_stack
        // or changes_undone_stack, as appropriate:
        for prior_change in &history[1..] {
            let prior_action = ChangeAction::from_str(&prior_change.get_string("action")?)?;
            tracing::debug!("The change prior to it is a {prior_action}: {prior_change:?}.");
            match action_to_push {
                ChangeAction::Do => match prior_action {
                    ChangeAction::Do | ChangeAction::Redo => {
                        tracing::debug!(
                            "Pushing change {cid} to changes_done",
                            cid = change_to_push.get_string("change_id")?
                        );
                        changes_done_stack.push(change_to_push.clone());
                        change_to_push = prior_change;
                        action_to_push =
                            ChangeAction::from_str(&change_to_push.get_string("action")?)?;
                        do_redo_count = 1;
                        undo_count = 0;
                        tracing::debug!(
                            "The next change to push is a {action_to_push}: {change_to_push:?}"
                        );
                    }
                    ChangeAction::Undo => {
                        tracing::debug!(
                            "Pushing change {cid} to changes_done",
                            cid = change_to_push.get_string("change_id")?
                        );
                        changes_done_stack.push(change_to_push.clone());
                        change_to_push = prior_change;
                        action_to_push =
                            ChangeAction::from_str(&change_to_push.get_string("action")?)?;
                        do_redo_count = 0;
                        undo_count = 1;
                        tracing::debug!(
                            "The next change to push is a {action_to_push}: {change_to_push:?}"
                        );
                    }
                },
                ChangeAction::Undo => match prior_action {
                    ChangeAction::Undo => {
                        if do_redo_count == 0 {
                            tracing::debug!(
                                "Pushing change {cid} to changes_undone",
                                cid = change_to_push.get_string("change_id")?
                            );
                            changes_undone_stack.push(change_to_push.clone());
                            change_to_push = prior_change;
                            action_to_push =
                                ChangeAction::from_str(&change_to_push.get_string("action")?)?;
                            tracing::debug!(
                                "The next change to push is a {action_to_push}: {change_to_push:?}"
                            );
                            undo_count += 1;
                        } else {
                            do_redo_count -= 1;
                            undo_count += 1;
                        }
                    }
                    ChangeAction::Do | ChangeAction::Redo => {
                        if undo_count == 0 {
                            tracing::debug!(
                                "Pushing change {cid} to changes_undone",
                                cid = change_to_push.get_string("change_id")?
                            );
                            changes_undone_stack.push(change_to_push.clone());
                            change_to_push = prior_change;
                            action_to_push =
                                ChangeAction::from_str(&change_to_push.get_string("action")?)?;
                            do_redo_count = 1;
                            tracing::debug!(
                                "The next change to push is a {action_to_push}: {change_to_push:?}"
                            );
                        } else {
                            do_redo_count += 1;
                            undo_count -= 1;
                        }
                    }
                },
                ChangeAction::Redo => match prior_action {
                    ChangeAction::Redo => {
                        if undo_count == 0 {
                            tracing::debug!(
                                "Pushing change {cid} to changes_done",
                                cid = change_to_push.get_string("change_id")?
                            );
                            changes_done_stack.push(change_to_push.clone());
                            change_to_push = prior_change;
                            action_to_push =
                                ChangeAction::from_str(&change_to_push.get_string("action")?)?;
                            tracing::debug!(
                                "The next change to push is a {action_to_push}: {change_to_push:?}"
                            );
                            do_redo_count += 1;
                        } else {
                            do_redo_count += 1;
                            undo_count -= 1;
                        }
                    }
                    ChangeAction::Do => {
                        if undo_count == 0 {
                            tracing::debug!(
                                "Pushing change {cid} to changes_done",
                                cid = change_to_push.get_string("change_id")?
                            );
                            changes_done_stack.push(change_to_push.clone());
                            change_to_push = prior_change;
                            action_to_push =
                                ChangeAction::from_str(&change_to_push.get_string("action")?)?;
                            tracing::debug!(
                                "The next change to push is a {action_to_push}: {change_to_push:?}"
                            );
                            // Dos begin anew.
                            do_redo_count = 1;
                        } else {
                            do_redo_count += 1;
                            undo_count -= 1;
                        }
                    }
                    ChangeAction::Undo => {
                        if do_redo_count == 0 {
                            tracing::debug!(
                                "Pushing change {cid} to changes_done",
                                cid = change_to_push.get_string("change_id")?
                            );
                            changes_done_stack.push(change_to_push.clone());
                            change_to_push = prior_change;
                            action_to_push =
                                ChangeAction::from_str(&change_to_push.get_string("action")?)?;
                            tracing::debug!(
                                "The next change to push is a {action_to_push}: {change_to_push:?}"
                            );
                            undo_count = 1;
                        } else {
                            do_redo_count -= 1;
                            undo_count += 1;
                        }
                    }
                },
            };

            tracing::debug!(
                "Updated the do/redo count to {do_redo_count}, and the undo count to \
                 {undo_count}."
            );

            // Remove duplicate entries from the stacks. These can result when a row is repeatedly
            // undone and redone. These are harmless, logically speaking, but they may potentially
            // confuse the user if they are included in the output.
            (changes_done_stack, changes_undone_stack) =
                prune_stacks(&changes_done_stack, &changes_undone_stack);

            // Check if we have exceeded the (max) context, and if so, stop looking for more
            // actions:
            let mut done_len = changes_done_stack.len();
            let mut undone_len = changes_undone_stack.len();
            match action_to_push {
                ChangeAction::Do | ChangeAction::Redo => done_len += 1,
                ChangeAction::Undo => undone_len += 1,
            };
            if let Some(context) = context {
                if done_len >= context && undone_len >= context {
                    break;
                }
            } else if done_len >= HISTORY_MAX || undone_len >= HISTORY_MAX {
                break;
            }
        }

        // Once we have finished iterating, there will be one action left over to push, which we
        // do now:
        match action_to_push {
            ChangeAction::Do | ChangeAction::Redo => {
                tracing::debug!("Pushing the last change to changes_done: {change_to_push:?}");
                changes_done_stack.push(change_to_push.clone());
            }
            ChangeAction::Undo => {
                tracing::debug!("Pushing the last change to changes_undone: {change_to_push:?}");
                changes_undone_stack.push(change_to_push.clone());
            }
        };

        // Don't return the contents of changes_undone if the last change was a do. Dos can never be
        // redone. If the last change was an undo or a redo, the logic will take care of itself and
        // it should never be possible to undo or redo inappropriately.
        let mut changes_undone_stack = match final_action {
            ChangeAction::Do => vec![],
            _ => changes_undone_stack,
        };

        // Prune the stacks one last time, in case by adding the final action we created a situation
        // in which one of the stacks contains duplicates:
        (changes_done_stack, changes_undone_stack) =
            prune_stacks(&changes_done_stack, &changes_undone_stack);

        // Similarly, crop for context one last time if a context has been defined:
        let history = match context {
            None => History {
                changes_done_stack,
                changes_undone_stack,
            },
            Some(context) => {
                let mut done_len = changes_done_stack.len();
                if done_len > context {
                    done_len = context;
                }
                let changes_done_stack = changes_done_stack[..done_len].to_vec();

                let mut undone_len = changes_undone_stack.len();
                if undone_len > context {
                    undone_len = context;
                }
                let changes_undone_stack = changes_undone_stack[..undone_len].to_vec();
                History {
                    changes_done_stack,
                    changes_undone_stack,
                }
            }
        };
        tracing::debug!("Returning history: {history:#?}");
        Ok(history)
    }

    /// Reverse the given changeset in the database
    async fn _revert(&self, change_id: u64, changeset: &ChangeSet) -> Result<Option<ChangeSet>> {
        tracing::trace!("Relatable::_revert({change_id}, {changeset:?})");
        match changeset.changes.first() {
            None => Ok(None),
            Some(change) => {
                if let Change::Update { .. } = change {
                    let conn = self.connection.reconnect()?;
                    let actual_changes = self._set_values(conn, &changeset).await?;
                    Ok(Some(actual_changes))
                } else {
                    let mut actual_changes = vec![];
                    for change in changeset.changes.iter() {
                        let conn = self.connection.reconnect()?;
                        match change {
                            Change::Update { .. } => (), // Change::Update already handled above.
                            Change::Add { row, after: _ } => {
                                let num_deleted = self
                                    ._delete_row(
                                        conn,
                                        &changeset.action,
                                        &changeset.table,
                                        &changeset.user,
                                        *row,
                                    )
                                    .await?;
                                if num_deleted > 0 {
                                    actual_changes.push(change.clone());
                                }
                            }
                            Change::Move {
                                row,
                                from_after,
                                to_after: _,
                            } => {
                                let new_order = self
                                    ._move_and_record_row(
                                        conn,
                                        &changeset.action,
                                        &changeset.table,
                                        &changeset.user,
                                        *row,
                                        *from_after,
                                    )
                                    .await?;
                                if new_order > 0 {
                                    actual_changes.push(change.clone());
                                }
                            }
                            Change::Delete { row, after } => {
                                // Get the row, as it was before it was deleted, from the history
                                // table:
                                let sql = format!(
                                    r#"SELECT "before" FROM "history"
                                       WHERE "change_id" = {sql_param}"#,
                                    sql_param = SqlParam::new(&self.connection.kind()).next()
                                );
                                let params = json!([change_id]);
                                let before = self
                                    .connection
                                    .query_one(&sql, Some(&params))
                                    .await?
                                    .ok_or(RelatableError::DataError(format!(
                                        "No history row found with change_id {change_id}"
                                    )))?
                                    .get_string("before")?;
                                let before = match serde_json::from_str::<JsonValue>(&before) {
                                    Err(err) => return Err(err.into()),
                                    Ok(JsonValue::Object(o)) => o,
                                    Ok(_) => {
                                        return Err(RelatableError::InputError(
                                            "The content parameter is not an object".to_string(),
                                        )
                                        .into());
                                    }
                                };
                                let before = JsonRow { content: before };
                                tracing::debug!(
                                    "Re-adding row '{before}' to table '{}'",
                                    changeset.table
                                );
                                self._add_row(
                                    conn,
                                    &changeset.action,
                                    &changeset.table,
                                    &changeset.user,
                                    Some(*row),
                                    Some(*after),
                                    &before,
                                )
                                .await?;
                                actual_changes.push(change.clone());
                            }
                        };
                    }
                    Ok(Some(ChangeSet {
                        action: changeset.action,
                        table: changeset.table.clone(),
                        user: changeset.user.clone(),
                        description: changeset.description.clone(),
                        changes: actual_changes,
                    }))
                }
            }
        }
    }

    /// Undo the last change made by the given user
    pub async fn undo(&self, user: &str) -> Result<Option<ChangeSet>> {
        tracing::trace!("Relatable::undo({user:?})");
        let (change_id, mut changeset) =
            match self.get_last_undoable_changeset_for_user(user).await? {
                None => {
                    tracing::warn!("Nothing to undo for '{user}'");
                    return Ok(None);
                }
                Some(changeset) => changeset,
            };
        changeset.action = ChangeAction::Undo;
        let changeset = self._revert(change_id, &changeset).await?;
        if let Some(_) = changeset {
            self.commit_to_git().await?;
        }
        Ok(changeset)
    }

    /// Redo the last change undone by the given user
    pub async fn redo(&self, user: &str) -> Result<Option<ChangeSet>> {
        tracing::trace!("Relatable::redo({user:?})");
        let (change_id, mut changeset) =
            match self.get_last_redoable_changeset_for_user(user).await? {
                None => {
                    tracing::warn!("Nothing to redo for '{user}'");
                    return Ok(None);
                }
                Some(changeset) => changeset,
            };
        tracing::debug!("Last redoable action (ID {change_id}) for user {user} was {changeset:?}");
        changeset.action = ChangeAction::Redo;
        let changeset = self._revert(change_id, &changeset).await?;
        if let Some(_) = changeset {
            self.commit_to_git().await?;
        }
        Ok(changeset)
    }

    /// Update the database using the given [ChangeSet]
    async fn _set_values(
        &self,
        mut conn: Option<DbActiveConnection>,
        changeset: &ChangeSet,
    ) -> Result<ChangeSet> {
        tracing::trace!("Relatable::set_values(conn, {changeset:?})");
        // Begin a transaction:
        let mut tx = self.connection.begin(&mut conn).await?;

        // Update the user cursor
        self.prepare_user_cursor(changeset, &mut tx)?;

        // Actually make the changes:
        let table = Table::_get_table(&changeset.table, &mut tx)?;
        let mut actual_changes = vec![];
        for change in &changeset.changes {
            match change {
                Change::Update {
                    row,
                    column,
                    before,
                    after,
                } => {
                    // Delete any existing messages associated with this row and column:
                    tracing::debug!(
                        "Deleting existing messages for column '{}.{}'",
                        table.name,
                        column
                    );
                    self._delete_message(
                        &mut tx,
                        &table.name,
                        Some(*row),
                        Some(column),
                        None,
                        None,
                    )?;

                    // Depending on whether this is an undo/redo or an original action, the
                    // new value will be taken from either `before` or `after`.
                    let before = JsonRow::nullify_value(&table, column, before);
                    let after = JsonRow::nullify_value(&table, column, after);
                    let mut cell = match &changeset.action {
                        ChangeAction::Undo | ChangeAction::Redo => Cell {
                            value: before.clone(),
                            text: sql::json_to_string(&before),
                            ..Default::default()
                        },
                        ChangeAction::Do => Cell {
                            value: after.clone(),
                            text: sql::json_to_string(&after),
                            ..Default::default()
                        },
                    };

                    // Validate the cell's SQL type and add any messages to the message table:
                    let column_config = table.get_config_for_column(column);
                    let mut sql_value = cell.value.clone();
                    if self.validation_level != ValidationLevel::None {
                        cell.validate_sql_type(&column_config)
                            .expect("Error validating cell");
                        for message in cell.messages.iter() {
                            let (msg_id, msg) = Relatable::_add_message(
                                "rltbl",
                                &table.name,
                                &row,
                                column,
                                &cell.value,
                                &message.level,
                                &message.rule,
                                &message.message,
                                &mut tx,
                            )?;
                            tracing::debug!("Added message (ID {msg_id}): {msg:?}");
                        }

                        // If the cell is invalid, insert a NULL instead of its actual value
                        if cell.has_sql_type_error() {
                            sql_value = JsonValue::Null;
                        }
                    }

                    // Generate the UPDATE statement:
                    let (sql, params) = {
                        let mut sql_param = SqlParam::new(&self.connection.kind());
                        let sql = format!(
                            r#"UPDATE "{table}"
                               SET "{column}" = {sql_value}
                               WHERE _id = {sql_param}
                               RETURNING 1 AS "updated""#,
                            table = changeset.table,
                            sql_value = match sql_value {
                                JsonValue::Null => "NULL".to_string(),
                                _ => sql_param.next(),
                            },
                            sql_param = sql_param.next()
                        );
                        let params = match sql_value {
                            JsonValue::Null => json!([row]),
                            _ => json!([sql_value, row]),
                        };
                        (sql, params)
                    };

                    tracing::debug!(
                        "Updating value of row {row} in {table}.{column} to {sql_value:?}",
                        table = table.name
                    );

                    // Execute the UPDATE statement.
                    if tx.query(&sql, Some(&params))?.len() < 1 {
                        tracing::warn!("No row with _id {row} found to update");
                    } else {
                        actual_changes.push(Change::Update {
                            row: *row,
                            column: column.clone(),
                            before: match &changeset.action {
                                ChangeAction::Undo | ChangeAction::Redo => after.clone(),
                                ChangeAction::Do => before.clone(),
                            },
                            after: match &changeset.action {
                                ChangeAction::Undo | ChangeAction::Redo => before.clone(),
                                ChangeAction::Do => after.clone(),
                            },
                        });
                    }

                    // Optionally do full validation on the newly updated cell and add further
                    // messages to the message table:
                    if self.validation_level == ValidationLevel::Full {
                        self._validate_column_optionally_for_row(
                            &column_config,
                            Some(row),
                            &mut tx,
                        )?;
                        for column in &column_config._get_dependent_columns(&mut tx)? {
                            tracing::debug!("Validating dependent column '{}'", column.name);
                            self._validate_structure_for_column_and_optionally_for_row(
                                column, None, &mut tx,
                            )?;
                        }
                    }
                }
                _ => {
                    return Err(RelatableError::InputError(format!(
                        "Invalid change in changeset argument to set_values(): {change:?}"
                    ))
                    .into());
                }
            };
        }

        let num_changes = actual_changes.len();
        let actual_changeset = ChangeSet {
            action: changeset.action,
            table: changeset.table.clone(),
            user: changeset.user.clone(),
            description: changeset.description.clone(),
            changes: actual_changes,
        };
        if num_changes > 0 {
            // Record the changes to the change and history tables:
            self.record_changeset(&actual_changeset, &mut tx)?;
        }

        // Commit the transaction:
        tx.commit()?;

        Ok(actual_changeset)
    }

    /// Update the database using the given [ChangeSet]
    pub async fn set_values(&self, changeset: &ChangeSet) -> Result<ChangeSet> {
        tracing::trace!("Relatable::set_values({changeset:?})");
        let conn = self.connection.reconnect()?;
        let changeset = self._set_values(conn, changeset).await?;
        if changeset.changes.len() > 0 {
            self.commit_to_git().await?;
        }
        Ok(changeset)
    }

    /// Add a message to the message table using the given [DbTransaction]
    pub fn _add_message(
        user: &str,
        table_name: &str,
        row: &u64,
        column: &str,
        value: &JsonValue,
        level: &str,
        rule: &str,
        message: &str,
        tx: &mut DbTransaction<'_>,
    ) -> Result<(u64, Message)> {
        tracing::trace!(
            "Relatable::add_message({user:?}, {table_name:?}, {row}, \
             {column:?}, {value:?}, {level:?}, {rule:?}, {message:?}, tx)"
        );

        let sql = format!(
            r#"INSERT INTO "message"
               ("added_by", "table", "row", "column", "value",
                "level", "rule", "message")
               VALUES
               ({sql_params})
               RETURNING "message_id""#,
            sql_params = SqlParam::new(&tx.kind()).get_as_list(8)
        );
        let params = json!([user, table_name, row, column, value, level, rule, message]);
        let message_id = tx
            .query_one(&sql, Some(&params))?
            .ok_or(RelatableError::DataError(
                "Error inserting message".to_string(),
            ))?
            .get_unsigned("message_id")?;

        Ok((
            message_id,
            Message {
                value: value.clone(),
                level: level.to_string(),
                rule: rule.to_string(),
                message: message.to_string(),
            },
        ))
    }

    /// Add a message to the message table.
    pub async fn add_message(
        &self,
        user: &str,
        table_name: &str,
        row: u64,
        column: &str,
        value: &JsonValue,
        level: &str,
        rule: &str,
        message: &str,
    ) -> Result<(u64, Message)> {
        tracing::trace!(
            "Relatable::add_message({self:?},  {user:?}, {table_name:?}, {row}, \
             {column:?}, {value:?}, {level:?}, {rule:?}, {message:?})"
        );

        // Begin a transaction:
        let mut conn = self.connection.reconnect()?;
        let mut tx = self.connection.begin(&mut conn).await?;

        let (message_id, message) = Relatable::_add_message(
            user, table_name, &row, column, value, level, rule, message, &mut tx,
        )?;

        // Commit the transaction:
        tx.commit()?;

        Ok((message_id, message))
    }

    /// Add a row to the given table
    async fn _add_row(
        &self,
        mut conn: Option<DbActiveConnection>,
        action: &ChangeAction,
        table_name: &str,
        user: &str,
        new_row_id: Option<u64>,
        after_id: Option<u64>,
        row: &JsonRow,
    ) -> Result<Row> {
        tracing::trace!(
            "Relatable::_add_row(conn, {action:?}, {user:?}, {new_row_id:?}, \
                         {after_id:?}, {row:?})"
        );

        // Begin a transaction:
        let mut tx = self.connection.begin(&mut conn).await?;

        // Get the current database information for the table:
        let table = Table::_get_table(table_name, &mut tx)?;
        if !table.editable {
            return Err(
                RelatableError::InputError(format!("{} is not editable.", table_name,)).into(),
            );
        }

        // Nullify the JSON row by setting any column values whose content matches the column's
        // nulltype to Null:
        let row = JsonRow::nullify(row, &table);

        // Prepare a new row to be inserted using the JSON row as a base:
        let mut new_row = Row::prepare_new(&table, Some(&row), &mut tx)?;

        // A new_row_id will have been passed if the row is being added as part of an undo/redo.
        // In that case an after_id must have been passed as well but we leave the row order as
        // is for now, since we are not assured that the old row order is actually still free in
        // the table (recall that there is a unique constraint on _order). However the row_order
        // currently assigned is at the end of the table so there should not be any conflicts.
        if let Some(new_row_id) = new_row_id {
            tracing::debug!("Changing new row ID to {new_row_id}");
            new_row.id = new_row_id;
        }

        // Validate the row and add it to the table:
        if self.validation_level != ValidationLevel::None {
            new_row.validate_sql_types(&table, &mut tx)?;
            for (_column, cell) in new_row.cells.iter_mut() {
                if cell.has_sql_type_error() {
                    cell.value = JsonValue::Null;
                    cell.text = "".to_string(); // Should it be "null" instead of blank?
                }
            }
        }
        let (sql, params) = new_row.as_insert(&table.name, &tx.kind());
        tx.query(&sql, Some(&params))?;

        // Optionally do full validation on the row after it has been inserted:
        if self.validation_level == ValidationLevel::Full {
            self._validate_row(&table, &new_row.id, &mut tx)?;
            for table in &table._get_dependent_tables(None, &mut tx)? {
                tracing::debug!("Validating dependent table '{}'", table.name);
                self._validate_structure_for_table(table, &mut tx)?;
            }
        }

        let after_id = match after_id {
            None => Table::_get_previous_row_id(&table.name, new_row.id, &mut tx)?,
            Some(after_id) => {
                // Move the row to its assigned spot within the table:
                tracing::debug!(
                    "Moving new row {id} to after row {after_id} in '{table}'",
                    id = new_row.id,
                    table = table.name
                );
                let new_order = self._move_row(&mut tx, &table, new_row.id, after_id)?;
                new_row.order = new_order;
                after_id
            }
        };

        tracing::debug!(
            "Added new row {id} to table '{table}' after row {after_id}",
            id = new_row.id,
            table = table.name
        );

        // Prepare a changeset to be recorded, consisting of a single change record indicating
        // the addition of one new row with the new_row's id and position in the table:
        let changeset = ChangeSet {
            action: *action,
            table: table_name.to_string(),
            user: user.to_string(),
            description: "Add one row".to_string(),
            changes: vec![Change::Add {
                row: new_row.id,
                after: after_id,
            }],
        };

        // Use the changeset to prepare the user cursor:
        self.prepare_user_cursor(&changeset, &mut tx)?;

        // Record the changes to the history table:
        self.record_changeset(&changeset, &mut tx)?;

        // Commit the transaction:
        tx.commit()?;

        Ok(new_row)
    }

    /// Add a row to the given table
    pub async fn add_row(
        &self,
        table_name: &str,
        user: &str,
        after_id: Option<u64>,
        row: &JsonRow,
    ) -> Result<Row> {
        tracing::trace!("Relatable::add_row({table_name:?}, {user:?}, {after_id:?}, {row:?})");
        let conn = self.connection.reconnect()?;
        let new_row = self
            ._add_row(
                conn,
                &ChangeAction::Do,
                table_name,
                user,
                None,
                after_id,
                row,
            )
            .await?;
        self.commit_to_git().await?;
        Ok(new_row)
    }

    /// Delete a row from the table. Returns the number of rows deleted.
    async fn _delete_row(
        &self,
        mut conn: Option<DbActiveConnection>,
        action: &ChangeAction,
        table_name: &str,
        user: &str,
        row: u64,
    ) -> Result<usize> {
        tracing::trace!(
            "Relatable::_delete_row(conn, {action:?}, {table_name:?}, {user:?} \
                         {row})"
        );
        // Begin a transaction:
        let mut tx = self.connection.begin(&mut conn).await?;

        // Get the current database information for the table:
        let table = Table::_get_table(table_name, &mut tx)?;
        if !table.editable {
            return Err(
                RelatableError::InputError(format!("{} is not editable.", table_name,)).into(),
            );
        }

        // Prepare a changeset to be recorded, consisting of a single change record indicating
        // that a row with the given row number at the given table position has been deleted:
        let changeset = ChangeSet {
            action: *action,
            table: table_name.to_string(),
            user: user.to_string(),
            description: "Delete one row".to_string(),
            changes: vec![Change::Delete {
                row: row,
                after: Table::_get_previous_row_id(table_name, row, &mut tx)?,
            }],
        };

        // Use the changeset to prepare the user cursor:
        self.prepare_user_cursor(&changeset, &mut tx)?;

        // Delete the row:
        let sql = format!(
            r#"DELETE FROM "{}" WHERE "_id" = {sql_param} RETURNING 1 AS "deleted""#,
            table.name,
            sql_param = SqlParam::new(&self.connection.kind()).next()
        );
        let params = json!([row]);
        tracing::debug!("Deleted row {row} from table {table_name}");

        // Delete any messages associated with the row
        self._delete_message(&mut tx, table_name, Some(row), None, None, None)?;
        tracing::debug!("Deleted messages for deleted row {row} of table {table_name}");

        // Record the change to the history table:
        self.record_changeset(&changeset, &mut tx)?;

        let num_deleted = tx.query(&sql, Some(&params))?.len();
        if num_deleted < 1 {
            tracing::warn!("No row found with _id {row} to delete");
            // Roll back the changes to the history and change table. The reason we made these
            // prior to the actual delete was so that we could record the row's position in the
            // table before it was deleted.
            tx.rollback()?;
        } else {
            // Commit the transaction:
            tx.commit()?;
        }

        Ok(num_deleted)
    }

    /// Delete a row from a given table
    pub async fn delete_row(&self, table_name: &str, user: &str, row: u64) -> Result<usize> {
        tracing::trace!("Relatable::delete_row({table_name:?}, {user:?}, {row})");
        let conn = self.connection.reconnect()?;
        let num_deleted = self
            ._delete_row(conn, &ChangeAction::Do, table_name, user, row)
            .await?;
        if num_deleted > 0 {
            self.commit_to_git().await?;
        }
        Ok(num_deleted)
    }

    /// Delete messages from the message table. Returns the number of messages deleted.
    pub async fn delete_message(
        &self,
        table: &str,
        row: Option<u64>,
        column: Option<&str>,
        target_rule: Option<&str>,
        target_user: Option<&str>,
    ) -> Result<usize> {
        tracing::trace!(
            "Relatable::delete_message({self:?}, {table:?}, {row:?}, {column:?}, \
             {target_rule:?}, {target_user:?})"
        );

        // Begin a transaction:
        let mut conn = self.connection.reconnect()?;
        let mut tx = self.connection.begin(&mut conn).await?;

        // Delete the messages using the transaction
        let num_deleted =
            self._delete_message(&mut tx, table, row, column, target_rule, target_user)?;

        // Commit the transaction:
        tx.commit()?;

        Ok(num_deleted)
    }

    /// Delete messages from the message table using the given transaction. Returns the
    /// number of messages deleted.
    fn _delete_message(
        &self,
        tx: &mut DbTransaction<'_>,
        table: &str,
        row: Option<u64>,
        column: Option<&str>,
        target_rule: Option<&str>,
        target_user: Option<&str>,
    ) -> Result<usize> {
        tracing::trace!(
            "Relatable::_delete_message({self:?}, tx, {table:?}, {row:?}, {column:?}, \
             {target_rule:?}, {target_user:?})"
        );

        let mut sql_param = SqlParam::new(&self.connection.kind());
        let mut sql = format!(
            r#"DELETE FROM "message" WHERE "table" = {sql_param}"#,
            sql_param = sql_param.next()
        );
        let mut params = vec![json!(table)];

        if let Some(row) = row {
            sql.push_str(&format!(
                r#" AND "row" = {sql_param}"#,
                sql_param = sql_param.next(),
            ));
            params.push(json!(row));
        }
        if let Some(column) = column {
            sql.push_str(&format!(
                r#" AND "column" = {sql_param}"#,
                sql_param = sql_param.next()
            ));
            params.push(json!(column));
        }
        if let Some(target_rule) = target_rule {
            sql.push_str(&format!(
                r#" AND "rule" LIKE {sql_param}"#,
                sql_param = sql_param.next()
            ));
            params.push(json!(target_rule));
        }
        if let Some(target_user) = target_user {
            sql.push_str(&format!(
                r#" AND "added_by" = {sql_param}"#,
                sql_param = sql_param.next()
            ));
            params.push(json!(target_user));
        }

        sql.push_str(r#" RETURNING 1 AS "deleted""#);
        let num_deleted = tx.query(&sql, Some(&json!(params)))?.len();
        Ok(num_deleted)
    }

    /// Move a row and record the change in the change table
    async fn _move_and_record_row(
        &self,
        mut conn: Option<DbActiveConnection>,
        action: &ChangeAction,
        table_name: &str,
        user: &str,
        id: u64,
        after_id: u64,
    ) -> Result<u64> {
        tracing::trace!(
            "Relatable::_move_and_record_row(conn, {action:?}, {table_name:?}, \
                         {user:?}, {id}, {after_id})"
        );

        // Begin a transaction:
        let mut tx = self.connection.begin(&mut conn).await?;

        // Get the current database information for the table:
        let table = Table::_get_table(table_name, &mut tx)?;
        if !table.editable {
            return Err(
                RelatableError::InputError(format!("{} is not editable.", table_name,)).into(),
            );
        }

        // Prepare a changeset to be recorded, consisting of a single change record indicating
        // that a row has been displaced from somewhere to somewhere else.
        let changeset = ChangeSet {
            action: *action,
            table: table_name.to_string(),
            user: user.to_string(),
            description: "Move one row".to_string(),
            changes: vec![Change::Move {
                row: id,
                from_after: Table::_get_previous_row_id(table_name, id, &mut tx)?,
                to_after: after_id,
            }],
        };

        // Use the changeset to prepare the user cursor:
        self.prepare_user_cursor(&changeset, &mut tx)?;

        // Move the row within the table:
        let new_order = self._move_row(&mut tx, &table, id, after_id)?;

        if new_order != 0 {
            // Record the change to the history table:
            self.record_changeset(&changeset, &mut tx)?;
        }

        // Commit the transaction:
        tx.commit()?;

        Ok(new_order)
    }

    /// Move a row to a different position in a given table
    fn _move_row(
        &self,
        tx: &mut DbTransaction<'_>,
        table: &Table,
        id: u64,
        after_id: u64,
    ) -> Result<u64> {
        tracing::trace!("Relatable::_move_row(tx, {table:?}, {id}, {after_id})");
        fn get_row_order(tx: &mut DbTransaction<'_>, table: &Table, row_id: u64) -> Result<u64> {
            let sql = format!(
                r#"SELECT "_order" FROM "{}" WHERE "_id" = {sql_param}"#,
                table.name,
                sql_param = SqlParam::new(&tx.kind()).next()
            );
            let params = json!([row_id]);
            let rows = tx.query(&sql, Some(&params))?;
            if rows.is_empty() {
                return Err(RelatableError::DataError(format!(
                    "Unable to fetch _order for row {row_id} of table '{table}'",
                    table = table.name
                ))
                .into());
            }
            match rows[0].content.get("_order").and_then(|o| o.as_u64()) {
                Some(order) => Ok(order as u64),
                None => {
                    return Err(
                        RelatableError::DataError("No integer '_order' in row".to_string()).into(),
                    )
                }
            }
        }

        // Get the order, (A), of `after_id`:
        let order_prev = {
            if after_id > 0 {
                let mut id_to_try = after_id;
                let mut result = get_row_order(tx, table, id_to_try);
                // This handles the case in which the after row has been deleted for some reason
                // (this might happen if we are redoing).
                while let Err(_) = result {
                    if id_to_try == 0 {
                        break;
                    }
                    tracing::debug!("Could not obtain _order for row {id_to_try}");
                    id_to_try -= 1;
                    tracing::debug!("Trying to find the _order of row {id_to_try}");
                    result = get_row_order(tx, table, id_to_try);
                }
                result?
            } else {
                // It is not possible for a row to be assigned a order of zero. We allow it as a
                // possible value of `after_id`, however, which is used as a special value that we
                // should move the row identified by `id` to the beginning of the table.
                0
            }
        };

        // Run a query to get the minimum order, (B), that is greater than (A).
        let order_next = {
            let sql = format!(
                r#"SELECT MIN("_order") AS "_order" FROM "{}" WHERE "_order" > {sql_param}"#,
                table.name,
                sql_param = SqlParam::new(&tx.kind()).next()
            );
            let params = json!([order_prev]);
            let rows = tx.query(&sql, Some(&params))?;
            if rows.is_empty() {
                return Err(RelatableError::DataError(format!(
                    "Could not determine the minimum row order greater than {order_prev}"
                ))
                .into());
            }

            match rows[0].content.get("_order") {
                Some(value) => match value {
                    JsonValue::Null => {
                        // The row_order will be null if we ask Relatable to move a row to
                        // a position after the last row in the table.
                        order_prev + NEW_ORDER_MULTIPLIER as u64
                    }
                    _ => match value.as_u64() {
                        Some(order) => order as u64,
                        None => {
                            return Err(RelatableError::DataError(
                                "Field '_order' in row is not an integer".to_string(),
                            )
                            .into());
                        }
                    },
                },
                None => {
                    return Err(RelatableError::DataError("No '_order' in row".to_string()).into());
                }
            }
        };

        let mut new_order = {
            if order_prev + 1 < order_next {
                // If the next order is not occupied just use it:
                order_prev + 1
            } else {
                // Otherwise, get all the orders that need to be moved. We sort the results in
                // descending order so that when we later update each value, no duplicate key
                // violations will ensue:
                let upper_bound = (order_next as f32 / NEW_ORDER_MULTIPLIER as f32).ceil() as u64
                    * NEW_ORDER_MULTIPLIER as u64;
                let mut sql_param = SqlParam::new(&tx.kind());
                let sql = format!(
                    r#"SELECT "_order"
                         FROM "{}"
                        WHERE "_order" >= {sql_param_1} AND "_order" < {sql_param_2}
                     ORDER BY "_order" DESC"#,
                    table.name,
                    sql_param_1 = sql_param.next(),
                    sql_param_2 = sql_param.next()
                );
                let params = json!([order_next, upper_bound]);
                let rows = tx.query(&sql, Some(&params))?;
                if rows.is_empty() {
                    return Err(RelatableError::DataError(
                        "Could not determine the highest row order".to_string(),
                    )
                    .into());
                }
                let highest_order = match rows[0].content.get("_order").and_then(|o| o.as_u64()) {
                    Some(order) => order as u64,
                    None => {
                        return Err(RelatableError::DataError(
                            "No field '_order' in row or it is not an integer".to_string(),
                        )
                        .into())
                    }
                };
                if highest_order + 1 >= upper_bound {
                    // Return an error
                    return Err(RelatableError::DataError(format!(
                        "Impossible to move row {} after row {}: No more room",
                        id, after_id
                    ))
                    .into());
                }

                for row in rows {
                    let current_order = match row.content.get("_order").and_then(|o| o.as_u64()) {
                        Some(order) => order as u64,
                        None => {
                            return Err(RelatableError::DataError(
                                "No field '_order' in row or it is not an integer".to_string(),
                            )
                            .into())
                        }
                    };
                    let sql = format!(
                        r#"UPDATE "{}"
                              SET "_order" = "_order" + 1
                            WHERE "_order" = {sql_param}"#,
                        table.name,
                        sql_param = SqlParam::new(&tx.kind()).next()
                    );
                    let params = json!([current_order]);
                    tx.query(&sql, Some(&params))?;
                }
                // Now that we have made some room, we can use order_prev + 1,
                // which should no longer be occupied:
                order_prev + 1
            }
        };

        tracing::debug!(
            "Updating _order in table '{table}' for row {id} to {new_order}",
            table = table.name
        );

        let mut sql_param = SqlParam::new(&tx.kind());
        let sql = format!(
            r#"UPDATE "{}" SET "_order" = {sql_param_1}
               WHERE "_id" = {sql_param_2}
               RETURNING 1 AS "moved""#,
            table.name,
            sql_param_1 = sql_param.next(),
            sql_param_2 = sql_param.next(),
        );
        let params = json!([new_order, id]);
        if tx.query(&sql, Some(&params))?.len() < 1 {
            tracing::warn!("Now row with _id {id} found to move");
            // It is not possible for a row to have an order of zero. It is used here to
            // represent the case where no row was actually moved to the caller.
            new_order = 0;
        }
        Ok(new_order)
    }

    /// Change the _id of the given row in the given table.
    fn _change_row_id(
        &self,
        tx: &mut DbTransaction<'_>,
        table: &Table,
        id: u64,
        new_id: u64,
    ) -> Result<()> {
        tracing::trace!("Relatable::_change_row_id(tx, {table:?}, {id}, {new_id})");
        let mut sql_param = SqlParam::new(&tx.kind());
        let sql = format!(
            r#"UPDATE "{table}"
                  SET "_id" = {sql_param_1}, "_order" = {sql_param_2}
                WHERE "_id" = {sql_param_3}
            RETURNING "_id" AS "_id""#,
            table = table.name,
            sql_param_1 = sql_param.next(),
            sql_param_2 = sql_param.next(),
            sql_param_3 = sql_param.next(),
        );
        let params = json!([new_id, id, id * NEW_ORDER_MULTIPLIER as u64]);
        tx.query_one(&sql, Some(&params))?
            .ok_or(RelatableError::DataError(format!("No row with _id = {id}")))?
            .get_unsigned("_id")?;
        Ok(())
    }

    /// Move a row to a different position in a given table.
    pub async fn move_row(
        &self,
        table_name: &str,
        user: &str,
        id: u64,
        after_id: u64,
    ) -> Result<u64> {
        tracing::trace!("Relatable::move_row({table_name:?}, {user:?}, {after_id:?})");
        let conn = self.connection.reconnect()?;
        let new_order = self
            ._move_and_record_row(conn, &ChangeAction::Do, table_name, user, id, after_id)
            .await?;
        if new_order != 0 {
            self.commit_to_git().await?;
        }
        Ok(new_order)
    }

    /// Validate all of the data in the given database table
    pub async fn validate_table(&self, table: &Table) -> Result<()> {
        tracing::trace!("Relatable::validate_table({self:?}, {table:?})");

        // Reconnect and begin a transaction:
        let mut conn = self.connection.reconnect()?;
        let mut tx = self.connection.begin(&mut conn).await?;

        self._validate_table(table, &mut tx)?;

        // Commit the transaction
        tx.commit()?;

        tracing::info!("Validated table '{}'", table.name);
        Ok(())
    }

    /// Validate all of the data in the given database table using the given transaction
    fn _validate_table(&self, table: &Table, tx: &mut DbTransaction<'_>) -> Result<()> {
        tracing::trace!("Relatable::_validate_table({self:?}, {table:?}, tx)");

        // Validate each table column
        for (_, column) in table.columns.iter() {
            self._validate_column_optionally_for_row(column, None, tx)?;
        }

        tracing::debug!("Validated table '{}'", table.name);
        Ok(())
    }

    /// Do datatype validation on all of the data in the given database table
    pub async fn validate_datatype_for_table(&self, table: &Table) -> Result<()> {
        tracing::trace!("Relatable::validate_datatype_for_table({self:?}, {table:?})");

        // Reconnect and begin a transaction:
        let mut conn = self.connection.reconnect()?;
        let mut tx = self.connection.begin(&mut conn).await?;

        self._validate_datatype_for_table(table, &mut tx)?;

        // Commit the transaction
        tx.commit()?;

        tracing::info!("Validated datatype for table '{}'", table.name);
        Ok(())
    }

    /// Do datatype validation on all of the data in the given table using the given database
    /// transaction
    fn _validate_datatype_for_table(
        &self,
        table: &Table,
        tx: &mut DbTransaction<'_>,
    ) -> Result<()> {
        tracing::trace!("Relatable::_validate_datatype_for_table({self:?}, {table:?}, tx)");

        // Validate each table column
        for (_, column) in table.columns.iter() {
            self._validate_datatype_for_column_and_optionally_for_row(column, None, tx)?;
        }

        tracing::debug!("Validated datatype for table '{}'", table.name);
        Ok(())
    }

    /// Do structure validation on all of the data in the given database table
    pub async fn validate_structure_for_table(&self, table: &Table) -> Result<()> {
        tracing::trace!("Relatable::validate_structure_for_table({self:?}, {table:?})");

        // Reconnect and begin a transaction:
        let mut conn = self.connection.reconnect()?;
        let mut tx = self.connection.begin(&mut conn).await?;

        self._validate_structure_for_table(table, &mut tx)?;

        // Commit the transaction
        tx.commit()?;

        tracing::info!("Validated structure for table '{}'", table.name);
        Ok(())
    }

    /// Do structure validation on all of the data in the given database table using the given
    /// database transation
    fn _validate_structure_for_table(
        &self,
        table: &Table,
        tx: &mut DbTransaction<'_>,
    ) -> Result<()> {
        tracing::trace!("Relatable::_validate_structure_for_table({self:?}, {table:?}, tx)");

        // Validate each table column
        for (_, column) in table.columns.iter() {
            self._validate_structure_for_column_and_optionally_for_row(column, None, tx)?;
        }

        tracing::debug!("Validated structure for table '{}'", table.name);
        Ok(())
    }

    /// Validate the data in the given column associated with a table in the database
    pub async fn validate_column(&self, column: &Column) -> Result<()> {
        tracing::trace!("Relatable::validate_column({self:?}, {column:?})");
        let mut conn = self.connection.reconnect()?;
        let mut tx = self.connection.begin(&mut conn).await?;
        self._validate_column_optionally_for_row(column, None, &mut tx)?;
        tx.commit()?;
        tracing::info!("Validated column '{}.{}'", column.table, column.name);
        Ok(())
    }

    /// Validate the value of the given column in the given row in the associated database
    /// table
    pub async fn validate_value(&self, column: &Column, row: &u64) -> Result<()> {
        tracing::trace!("Relatable::validate_value({self:?}, {column:?}, {row})");
        let mut conn = self.connection.reconnect()?;
        let mut tx = self.connection.begin(&mut conn).await?;
        self._validate_column_optionally_for_row(column, Some(row), &mut tx)?;
        tx.commit()?;
        tracing::info!(
            "Validated value at row {}, column '{}.{}'",
            row,
            column.table,
            column.name
        );
        Ok(())
    }

    /// Validate the given row of the given table
    pub async fn validate_row(&self, table: &Table, row: &u64) -> Result<()> {
        tracing::trace!("Relatable::validate_row({self:?}, {table:?}, {row})");
        let mut conn = self.connection.reconnect()?;
        let mut tx = self.connection.begin(&mut conn).await?;
        self._validate_row(table, row, &mut tx)?;
        tx.commit()?;
        tracing::info!("Validated row {} of table '{}'", row, table.name);
        Ok(())
    }

    /// Validate the given row of the given table using the given database transaction
    fn _validate_row(&self, table: &Table, row: &u64, tx: &mut DbTransaction<'_>) -> Result<()> {
        tracing::trace!("Relatable::_validate_row({self:?}, {table:?}, {row}, tx)");
        for (_, column) in table.columns.iter() {
            self._validate_column_optionally_for_row(column, Some(row), tx)?;
        }
        tracing::debug!("Validated row {} of table '{}'", row, table.name);
        Ok(())
    }

    /// Validate the datatype of the given column in its associated database table using the
    /// given transaction. If `row` is given, only validate the column for that row.
    fn _validate_datatype_for_column_and_optionally_for_row(
        &self,
        column: &Column,
        row: Option<&u64>,
        tx: &mut DbTransaction<'_>,
    ) -> Result<()> {
        tracing::trace!(
            "Relatable::_validate_datatype_for_column_and_optionally_for_row(\
             {self:?}, {column:?}, {row:?}, tx)"
        );

        let table_name = column.table.as_str();

        // Delete pre-existing datatype validation messages for this column and then
        // validate the datatype conditions for each datatype in the column's datatype hierarchy.
        self._delete_message(
            tx,
            table_name,
            row.copied(),
            Some(&column.name),
            Some("datatype:%"),
            Some("rltbl"),
        )?;

        // Gather the datatypes to check: The column's datatype, plus any further datatypes in
        // the datatype hierarchy:
        let mut datatypes_to_check = vec![column.datatype.clone()];
        datatypes_to_check.append(&mut column.datatype_hierarchy.clone());

        // Validate the column against each datatype in the hierarchy:
        for datatype in datatypes_to_check {
            let inserted = datatype.validate(column, row, tx)?;
            if !inserted {
                break;
            }
        }

        tracing::debug!(
            "Validated datatype for column: '{}.{}'{}",
            column.table,
            column.name,
            match row {
                None => "".to_string(),
                Some(row) => format!(", row: {row}"),
            }
        );
        Ok(())
    }

    /// Validate the structure of the given column in its associated database table using the
    /// given transaction. If `row` is given, only validate the column for that row.
    fn _validate_structure_for_column_and_optionally_for_row(
        &self,
        column: &Column,
        row: Option<&u64>,
        tx: &mut DbTransaction<'_>,
    ) -> Result<()> {
        tracing::trace!(
            "Relatable::_validate_structure_for_column_and_optionally_for_row(\
             {self:?}, {column:?}, {row:?}, tx)"
        );

        let table_name = column.table.as_str();

        // Delete pre-existing structure validation messages for this column and then re-validate
        // the structure condition for this column and (optionally) row:
        self._delete_message(
            tx,
            table_name,
            row.copied(),
            Some(&column.name),
            Some("key:%"),
            Some("rltbl"),
        )?;

        // Validate the cell's structure condition:
        if let Some(structure) = &column.structure {
            structure.validate(column, row, tx)?;
        }

        tracing::debug!(
            "Validated structure for column: '{}.{}'{}",
            column.table,
            column.name,
            match row {
                None => "".to_string(),
                Some(row) => format!(", row: {row}"),
            }
        );
        Ok(())
    }

    /// Validate the given column in its associated database table using the given transaction.
    /// If `row` is given, only validate the column for that row.
    fn _validate_column_optionally_for_row(
        &self,
        column: &Column,
        row: Option<&u64>,
        tx: &mut DbTransaction<'_>,
    ) -> Result<()> {
        tracing::trace!(
            "Relatable::_validate_column_optionally_for_row({self:?}, {column:?}, {row:?}, tx)"
        );
        self._validate_datatype_for_column_and_optionally_for_row(column, row, tx)?;
        self._validate_structure_for_column_and_optionally_for_row(column, row, tx)?;
        tracing::debug!(
            "Validated column: '{}.{}'{}",
            column.table,
            column.name,
            match row {
                None => "".to_string(),
                Some(row) => format!(", row: {row}"),
            }
        );
        Ok(())
    }

    /// Delete all entries from the cache corresponding to the given table, or clear it completely
    /// if no table is given.
    pub(crate) fn clear_cache(tx: &mut DbTransaction<'_>, table: Option<&str>) -> Result<()> {
        let mut sql = r#"DELETE FROM "cache""#.to_string();
        if let Some(table) = table {
            let mut table = table.to_string();
            tracing::debug!("Deleting entries for table '{table}' from cache");
            match tx.kind() {
                DbKind::Postgres => {
                    // Note that the '?' is *not* being used as a parameter placeholder here
                    // but a JSONB operator.
                    sql.push_str(&format!(
                        r#" WHERE "tables" ? {}"#,
                        SqlParam::new(&tx.kind()).next()
                    ));
                }
                DbKind::Sqlite => {
                    sql.push_str(&format!(
                        r#" WHERE "tables" LIKE {}"#,
                        SqlParam::new(&tx.kind()).next()
                    ));
                    table = format!(r#"%"{table}"%"#);
                }
            };
            let params = json!([table]);
            tx.query(&sql, Some(&params))?;
        } else {
            tracing::debug!("Truncating cache");
            tx.query(&sql, None)?;
        }

        Ok(())
    }

    /// Delete all entries from the in-memory cache corresponding to the given table
    pub(crate) fn clear_mem_cache(&self, table: &str) {
        let table = format!("\"{table}\"");
        let mut cache = CACHE.lock().expect("Could not lock cache");
        let keys = cache
            .keys()
            .map(|k| k)
            .cloned()
            .collect::<HashSet<_>>()
            .into_iter()
            .collect::<Vec<_>>();
        for key in keys.iter() {
            if key.tables.contains(&table) {
                tracing::debug!("Removing {key:?} from cache");
                cache.remove(key);
            }
        }
    }
}

// Validation

/// The level at which Relatable will perform validation when adding to or modifying data in the
/// database
#[derive(Clone, Copy, Debug, PartialEq, Serialize, Deserialize)]
pub enum ValidationLevel {
    /// Perform no validateion
    None,
    /// Perform only SQL type validation
    SqlType,
    /// Perform full validation
    Full,
}

impl FromStr for ValidationLevel {
    type Err = anyhow::Error;

    fn from_str(level: &str) -> Result<Self> {
        tracing::trace!("ValidationLevel::from_str({level:?})");
        match level.to_lowercase().as_str() {
            "none" => Ok(ValidationLevel::None),
            "sql_type" => Ok(ValidationLevel::SqlType),
            "full" => Ok(ValidationLevel::Full),
            _ => {
                return Err(
                    RelatableError::InputError(format!("Unrecognized level: {level}")).into(),
                );
            }
        }
    }
}

// Changes and History

/// A set of changes made by a user to a table.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct ChangeSet {
    pub action: ChangeAction,
    pub table: String,
    pub user: String,
    pub description: String,
    pub changes: Vec<Change>,
}

impl ChangeSet {
    /// Given a change, returns the a [Cursor] representing where the user's cursor
    /// should be placed in the frontend.
    fn to_cursor(&self) -> Result<Cursor> {
        tracing::trace!("ChangeSet::to_cursor()");
        let table = self.table.clone();
        match self.changes.first() {
            Some(change) => match change {
                Change::Update {
                    row,
                    column,
                    before: _,
                    after: _,
                } => Ok(Cursor {
                    table,
                    row: *row,
                    column: column.to_string(),
                }),
                Change::Add { row, after: _ } => Ok(Cursor {
                    table,
                    row: *row,
                    column: "".to_string(),
                }),
                Change::Move {
                    row,
                    from_after: _,
                    to_after: _,
                } => Ok(Cursor {
                    table,
                    row: *row,
                    column: "".to_string(),
                }),
                Change::Delete { row, after: _ } => Ok(Cursor {
                    table,
                    row: *row,
                    column: "".to_string(),
                }),
            },
            None => Err(RelatableError::ChangeError("No changes in set".into()).into()),
        }
    }
}

/// The kind of action that is performed by a change
#[derive(Clone, Copy, Debug, Serialize, Deserialize, PartialEq, Eq)]
pub enum ChangeAction {
    Do,
    Undo,
    Redo,
}

impl FromStr for ChangeAction {
    type Err = anyhow::Error;

    fn from_str(action: &str) -> Result<Self> {
        tracing::trace!("ChangeAction::from_str({action:?})");
        match action.to_lowercase().as_str() {
            "do" => Ok(ChangeAction::Do),
            "undo" => Ok(ChangeAction::Undo),
            "redo" => Ok(ChangeAction::Redo),
            _ => {
                return Err(
                    RelatableError::InputError(format!("Unrecognized action: {action}")).into(),
                );
            }
        }
    }
}

impl Display for ChangeAction {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            ChangeAction::Do => write!(f, "do"),
            ChangeAction::Undo => write!(f, "undo"),
            ChangeAction::Redo => write!(f, "redo"),
        }
    }
}

/// A change to a table in the database
#[derive(Clone, Debug, Serialize, Deserialize)]
#[serde(tag = "type")]
pub enum Change {
    Update {
        /// The id of the row that was updated
        row: u64,
        /// The column whose value was updated
        column: String,
        /// The value of the column before the change
        before: JsonValue,
        /// The value of the column after the change
        after: JsonValue,
    },
    Add {
        /// The id of the row that was added
        row: u64,
        /// The _id of the row whose _order this comes immediately after in the table
        after: u64,
    },
    Move {
        /// The id of the row that was moved
        row: u64,
        /// The row that this row came after before the change
        from_after: u64,
        /// The row that this row came after after the change
        to_after: u64,
    },
    Delete {
        /// The id of the row that was deleted
        row: u64,
        /// The _id of the row whose _order this row came immediately after in the table before
        /// being deleted.
        after: u64,
    },
}

impl Change {
    /// Converts a JSON string representing an array of changes to an array of [Change] structs.
    pub fn many_from_str(content: &str) -> Result<Vec<Self>> {
        tracing::trace!("Change::many_from_str({content:?})");
        let json_content = match serde_json::from_str::<JsonValue>(content) {
            Err(err) => return Err(err.into()),
            Ok(JsonValue::Array(v)) => v,
            Ok(_) => {
                return Err(RelatableError::InputError(
                    "The content parameter is not an array".to_string(),
                )
                .into());
            }
        };

        let mut changes = vec![];
        for change_json in json_content.iter() {
            let change_json = match change_json.as_object() {
                Some(change_object) => JsonRow {
                    content: change_object.clone(),
                },
                None => {
                    return Err(RelatableError::InputError(format!(
                        "Not an object: {change_json}"
                    ))
                    .into());
                }
            };

            let change_type = change_json.get_string("type")?;
            let row = change_json.get_unsigned("row")?;
            match change_type.as_str() {
                "Update" => changes.push(Change::Update {
                    row: row,
                    column: change_json.get_string("column")?,
                    before: change_json.get_value("before")?,
                    after: change_json.get_value("after")?,
                }),
                "Add" => changes.push(Change::Add {
                    row: row,
                    after: change_json.get_unsigned("after")?,
                }),
                "Delete" => changes.push(Change::Delete {
                    row: row,
                    after: change_json.get_unsigned("after")?,
                }),
                "Move" => changes.push(Change::Move {
                    row: row,
                    from_after: change_json.get_unsigned("from_after")?,
                    to_after: change_json.get_unsigned("to_after")?,
                }),
                _ => {
                    return Err(RelatableError::InputError(format!(
                        "Unrecognized change type for change: {change_json}"
                    ))
                    .into());
                }
            };
        }
        Ok(changes)
    }

    /// Convers a [JsonRow] to a [Change]
    pub fn from_json_row(json_row: &JsonRow) -> Result<Self> {
        tracing::trace!("Change::from_json_row({json_row:?})");
        match json_row.get_string("type")?.as_str() {
            "Update" => Ok(Change::Update {
                row: json_row.get_unsigned("row")?,
                column: json_row.get_string("column")?,
                before: json_row.get_value("before")?,
                after: json_row.get_value("after")?,
            }),
            "Add" => Ok(Change::Add {
                row: json_row.get_unsigned("row")?,
                after: json_row.get_unsigned("after")?,
            }),
            "Move" => Ok(Change::Move {
                row: json_row.get_unsigned("row")?,
                from_after: json_row.get_unsigned("from_after")?,
                to_after: json_row.get_unsigned("to_after")?,
            }),
            "Delete" => Ok(Change::Delete {
                row: json_row.get_unsigned("row")?,
                after: json_row.get_unsigned("after")?,
            }),
            _ => {
                return Err(RelatableError::InputError(format!(
                    "Unrecognized action type for change {json_row}"
                ))
                .into());
            }
        }
    }
}

/// Describes a history of changes that have been done and undone.
#[derive(Default, Debug, Serialize, Deserialize)]
pub struct History {
    pub changes_done_stack: Vec<JsonRow>,
    pub changes_undone_stack: Vec<JsonRow>,
}

impl Display for Change {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Change::Update {
                row,
                column,
                before,
                after,
            } => {
                write!(
                    f,
                    "Update '{column}' in row {row} from {before} to {after}",
                    before = sql::json_to_string(before),
                    after = sql::json_to_string(after)
                )
            }
            Change::Add { row, after } => {
                write!(f, "Add row {row} after row {after}")
            }
            Change::Move {
                row,
                from_after,
                to_after,
            } => {
                write!(
                    f,
                    "Move row {row} from after row {from_after} to after row {to_after}"
                )
            }
            Change::Delete { row, after: _ } => write!(f, "Delete row {row}"),
        }
    }
}

// Ranges and Results

#[derive(Clone, Debug, Default, Serialize, Deserialize)]
pub struct Range {
    count: usize,
    total: u64,
    start: u64,
    end: u64,
}

impl std::fmt::Display for Range {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "Rows {}-{} of {}", self.start, self.end, self.total)
    }
}

#[derive(Clone, Debug, Default, Serialize, Deserialize)]
pub struct ResultSet {
    pub select: Select,
    pub statement: String,
    pub parameters: Vec<JsonValue>,
    pub range: Range,
    pub table: Table,
    /// The columns (and only the columns) used in the Select statement
    pub columns: Vec<Column>,
    pub rows: Vec<Row>,
}

impl ResultSet {
    /// Write the result set to CSV
    pub fn to_csv(&self) -> String {
        let writer = WriterBuilder::new().from_writer(vec![]);
        self.to_xsv(writer)
    }

    /// Write the result set to TSV
    pub fn to_tsv(&self) -> String {
        let writer = WriterBuilder::new()
            .delimiter(b'\t')
            .quote_style(QuoteStyle::Never)
            .from_writer(vec![]);
        self.to_xsv(writer)
    }

    /// Write the result set to XSV
    pub fn to_xsv(&self, mut writer: Writer<Vec<u8>>) -> String {
        let header_row = &self
            .columns
            .iter()
            .map(|c| c.name.clone())
            .collect::<Vec<String>>();
        writer.write_record(header_row.clone()).unwrap();
        for row in &self.rows {
            writer.write_record(row.to_strings()).unwrap();
        }
        String::from_utf8(writer.into_inner().unwrap()).unwrap()
    }

    /// Uses the given (unverified) printf-style format string and the given compiled regular
    /// expression (which is used to verify the given format) to format the given cell.
    fn format_cell_text_value(column_format: &str, format_regex: &Regex, cell: &str) -> String {
        // If the cell is an empty string, just return it as is:
        if cell == "" {
            return "".to_string();
        }

        let conversion_spec = match format_regex.captures(column_format) {
            Some(c) => c[1].to_lowercase(),
            None => {
                tracing::warn!("Illegal format: '{}'", column_format);
                "s".to_string()
            }
        };
        let generic_error = format!("Error applying format '{}' to '{}'", column_format, cell);
        match conversion_spec.as_str() {
            "d" | "i" | "c" => match cell.parse::<isize>() {
                Ok(cell) => match sprintf!(&column_format, cell) {
                    Ok(cell) => {
                        // For some reason sprintf converts signed ints to unsigned ints before
                        // converting them to a string. So we have to workaround this here:
                        let cell = cell.parse::<usize>().unwrap();
                        let cell = cell as isize;
                        cell.to_string()
                    }
                    Err(e) => {
                        tracing::warn!("{}: {}", generic_error, e);
                        cell.to_string()
                    }
                },
                Err(e) => {
                    tracing::warn!("{}: {}", generic_error, e);
                    cell.to_string()
                }
            },
            "o" | "u" | "x" => match cell.parse::<usize>() {
                Ok(cell) => sprintf!(&column_format, cell).unwrap_or(cell.to_string()),
                Err(e) => {
                    tracing::warn!("{}: {}", generic_error, e);
                    cell.to_string()
                }
            },
            "e" | "f" | "g" | "a" => match cell.parse::<f64>() {
                Ok(cell) => sprintf!(&column_format, cell).unwrap_or(cell.to_string()),
                Err(e) => {
                    tracing::warn!("{}: {}", generic_error, e);
                    cell.to_string()
                }
            },
            "s" => sprintf!(&column_format, cell).unwrap_or(cell.to_string()),
            _ => {
                tracing::warn!(
                    "Unsupported conversion specifier '{}' in column format '{}'",
                    conversion_spec,
                    column_format
                );
                cell.to_string()
            }
        }
    }

    /// Write the result set to the console
    pub fn to_console(&self) -> String {
        let tw = TabWriter::new(vec![]);
        let mut tw = tw.ansi(true);
        tw.write(format!("{}\n", self.range).as_bytes())
            .unwrap_or_default();
        let header = &self
            .columns
            .iter()
            .map(|c| c.name.clone())
            .collect::<Vec<String>>();
        tw.write(format!("{}\n", header.join("\t")).as_bytes())
            .unwrap_or_default();

        let format_regex = Regex::new(r#"^%.*([\w%])$"#).expect("Invalid regular expression");
        let mut contains_errors = false;
        for row in &self.rows {
            let cells = row
                .cells
                .iter()
                .map(|(column_name, cell)| {
                    let value_to_print = {
                        let column_format = match self.table.columns.get(column_name) {
                            Some(column) if column.datatype.format == "" => "%s",
                            Some(column) => &column.datatype.format,
                            None => {
                                tracing::warn!(
                                    "Can't determine cell format. No column found: '{column_name}'"
                                );
                                "%s"
                            }
                        };
                        ResultSet::format_cell_text_value(&column_format, &format_regex, &cell.text)
                    };
                    if cell.message_level() >= 2 {
                        contains_errors = true;
                        format!("{}", value_to_print.red())
                    } else {
                        value_to_print
                    }
                })
                .collect::<Vec<_>>();
            tw.write(format!("{}\n", cells.join("\t")).as_bytes())
                .unwrap_or_default();
        }
        tw.flush().expect("TabWriter to flush");
        let written = String::from_utf8(tw.into_inner().unwrap()).unwrap();
        written
    }
}

impl std::fmt::Display for ResultSet {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let mut tw = TabWriter::new(vec![]);
        tw.write(format!("{}\n", self.range).as_bytes())
            .unwrap_or_default();
        let header = &self
            .columns
            .iter()
            .map(|c| c.name.clone())
            .collect::<Vec<String>>();
        tw.write(format!("{}\n", header.join("\t")).as_bytes())
            .unwrap_or_default();
        for row in &self.rows {
            tw.write(format!("{}\n", row.to_strings().join("\t")).as_bytes())
                .unwrap_or_default();
        }
        tw.flush().expect("TabWriter to flush");
        let written = String::from_utf8(tw.into_inner().unwrap()).unwrap();
        write!(f, "{written}")
    }
}

// Web Site Stuff

#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct Site {
    pub title: String,
    pub root: String,
    pub editable: bool,
    pub user: Account,
    pub users: IndexMap<String, UserCursor>,
    pub tables: Vec<String>,
}

#[derive(Clone, Debug, Default, Serialize, Deserialize)]
pub struct Account {
    name: String,
    color: String,
}

#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct Cursor {
    table: String,
    row: u64,
    column: String,
}

#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct UserCursor {
    name: String,
    color: String,
    cursor: Cursor,
    datetime: String,
}

#[derive(Clone, Debug, Default, Serialize, Deserialize)]
pub struct Page {
    pub path: String,
    pub formats: IndexMap<String, String>,
    pub tabs: Vec<Tab>,
}

#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct Tab {
    pub table: String,
    pub active: bool,
    pub url: String,
    pub count: String,
}