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;
pub static RLTBL_DEFAULT_DB: &str = ".relatable/relatable.db";
pub static NEW_ORDER_MULTIPLIER: usize = 1000;
pub static HISTORY_MAX: usize = 1000;
pub static DEFAULT_LIMIT: usize = 100;
pub static MAX_LIMIT: usize = 1000;
lazy_static! {
pub static ref CACHE: Mutex<HashMap<MemoryCacheKey, Vec<JsonRow>>> = Mutex::new(HashMap::new());
}
#[derive(Debug)]
pub enum RelatableError {
ChangeError(String),
ConfigError(String),
DataError(String),
GitError(String),
InitError(String),
FormatError(String),
InputError(String),
IOError(std::io::Error),
MissingError(String),
SerdeJsonError(serde_json::Error),
RegexError(regex::Error),
TableError(String),
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 {}
#[derive(Debug)]
pub struct Relatable {
pub root: String,
pub readonly: bool,
pub connection: DbConnection,
pub default_limit: usize,
pub max_limit: usize,
pub caching_strategy: CachingStrategy,
pub validation_level: ValidationLevel,
pub memory_cache_size: usize,
}
impl Relatable {
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();
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,
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,
},
})
}
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)?;
}
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)
}
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)
}
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",
};
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?;
}
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(¶m_values)).await?;
}
Ok(())
}
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",
};
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(())
}
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(¶m_values)).await?;
Ok(())
}
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(¶m_values)).await?;
Ok(())
}
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();
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();
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();
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(())
}
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(())
}
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(())
}
pub async fn drop_database(&self) -> Result<()> {
tracing::trace!("Relatable::drop_database({self:?})");
self.drop_data_tables().await?;
self.drop_meta_tables().await?;
Ok(())
}
pub fn render<T: Serialize>(&self, template: &str, context: T) -> Result<String> {
tracing::trace!("Relatable::render({template:?}, context)");
let mut env = Environment::new();
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")),
]);
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())
}
pub async fn fetch(&self, select: &Select) -> Result<ResultSet> {
tracing::trace!("Relatable::fetch({select:?})");
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<_>>();
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");
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();
}
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,
})
}
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(¶ms)).await
}
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(¶ms),
&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),
}
}
pub async fn load_table(&self, table_name: &str, path: &str, force: bool) {
tracing::trace!("Relatable::load_table({table_name:?}, {path:?}, {force})");
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();
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();
if force {
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(¶ms))
.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(¶ms))
.await
.expect("Error inserting to table table");
tracing::debug!("Table {table_name} (path: {path}) added to table table");
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
};
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");
}
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");
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) = {
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);
}
}
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()
}
};
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:?}");
}
}
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(", ")
};
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(¶m_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");
}
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
{
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(())
}
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}'");
self.save_all(None).await?;
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());
}
let (last_commit_author, days_ago) = git::get_last_commit_info()?;
let is_amendment = (last_commit_author == author) && (days_ago < 1);
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)?;
git::commit("commit by rltbl", &author, is_amendment)?;
Ok(())
}
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(¶ms))?;
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,
},
)))
}
}
}
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();
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,
};
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(¶ms))?;
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(¶ms))?;
}
Change::Add { row, after: _ } => {
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(¶ms))?
.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(¶ms))?;
}
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(¶ms))?;
}
Change::Delete { row, after: _ } => {
let json_row = match Table::_get_row(&table, *row, tx)? {
Some(json_row) => json_row,
None => {
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(¶ms))?;
}
};
}
match self.caching_strategy {
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(())
}
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()
}
}
}
}
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"#,
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)
}
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<_>>())
}
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()?;
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
};
tx.commit()?;
Ok(columns)
}
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()
}
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)
}
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(),
}
}
pub fn prepare_user_cursor(
&self,
changeset: &ChangeSet,
tx: &mut DbTransaction<'_>,
) -> Result<()> {
tracing::trace!("Relatable::prepare_user_cursor({changeset:?}, tx)");
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(¶ms))? {
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(¶ms))?;
}
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(¶ms))?;
Ok(())
}
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,
},
)))
}
}
}
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,
},
)))
}
}
}
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)
}
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(¶ms)).await?;
let mut changes_done_stack = vec![];
let mut changes_undone_stack = vec![];
let mut do_redo_count: usize;
let mut undo_count: usize;
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 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:?}"
);
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}."
);
(changes_done_stack, changes_undone_stack) =
prune_stacks(&changes_done_stack, &changes_undone_stack);
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;
}
}
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());
}
};
let mut changes_undone_stack = match final_action {
ChangeAction::Do => vec![],
_ => changes_undone_stack,
};
(changes_done_stack, changes_undone_stack) =
prune_stacks(&changes_done_stack, &changes_undone_stack);
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)
}
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::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 } => {
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(¶ms))
.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,
}))
}
}
}
}
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)
}
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)
}
async fn _set_values(
&self,
mut conn: Option<DbActiveConnection>,
changeset: &ChangeSet,
) -> Result<ChangeSet> {
tracing::trace!("Relatable::set_values(conn, {changeset:?})");
let mut tx = self.connection.begin(&mut conn).await?;
self.prepare_user_cursor(changeset, &mut tx)?;
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,
} => {
tracing::debug!(
"Deleting existing messages for column '{}.{}'",
table.name,
column
);
self._delete_message(
&mut tx,
&table.name,
Some(*row),
Some(column),
None,
None,
)?;
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()
},
};
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 cell.has_sql_type_error() {
sql_value = JsonValue::Null;
}
}
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
);
if tx.query(&sql, Some(¶ms))?.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(),
},
});
}
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 {
self.record_changeset(&actual_changeset, &mut tx)?;
}
tx.commit()?;
Ok(actual_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)
}
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(¶ms))?
.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(),
},
))
}
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:?})"
);
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,
)?;
tx.commit()?;
Ok((message_id, message))
}
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:?})"
);
let mut tx = self.connection.begin(&mut conn).await?;
let table = Table::_get_table(table_name, &mut tx)?;
if !table.editable {
return Err(
RelatableError::InputError(format!("{} is not editable.", table_name,)).into(),
);
}
let row = JsonRow::nullify(row, &table);
let mut new_row = Row::prepare_new(&table, Some(&row), &mut tx)?;
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;
}
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(); }
}
}
let (sql, params) = new_row.as_insert(&table.name, &tx.kind());
tx.query(&sql, Some(¶ms))?;
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) => {
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
);
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,
}],
};
self.prepare_user_cursor(&changeset, &mut tx)?;
self.record_changeset(&changeset, &mut tx)?;
tx.commit()?;
Ok(new_row)
}
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)
}
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})"
);
let mut tx = self.connection.begin(&mut conn).await?;
let table = Table::_get_table(table_name, &mut tx)?;
if !table.editable {
return Err(
RelatableError::InputError(format!("{} is not editable.", table_name,)).into(),
);
}
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)?,
}],
};
self.prepare_user_cursor(&changeset, &mut tx)?;
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}");
self._delete_message(&mut tx, table_name, Some(row), None, None, None)?;
tracing::debug!("Deleted messages for deleted row {row} of table {table_name}");
self.record_changeset(&changeset, &mut tx)?;
let num_deleted = tx.query(&sql, Some(¶ms))?.len();
if num_deleted < 1 {
tracing::warn!("No row found with _id {row} to delete");
tx.rollback()?;
} else {
tx.commit()?;
}
Ok(num_deleted)
}
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)
}
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:?})"
);
let mut conn = self.connection.reconnect()?;
let mut tx = self.connection.begin(&mut conn).await?;
let num_deleted =
self._delete_message(&mut tx, table, row, column, target_rule, target_user)?;
tx.commit()?;
Ok(num_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)
}
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})"
);
let mut tx = self.connection.begin(&mut conn).await?;
let table = Table::_get_table(table_name, &mut tx)?;
if !table.editable {
return Err(
RelatableError::InputError(format!("{} is not editable.", table_name,)).into(),
);
}
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,
}],
};
self.prepare_user_cursor(&changeset, &mut tx)?;
let new_order = self._move_row(&mut tx, &table, id, after_id)?;
if new_order != 0 {
self.record_changeset(&changeset, &mut tx)?;
}
tx.commit()?;
Ok(new_order)
}
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(¶ms))?;
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(),
)
}
}
}
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);
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 {
0
}
};
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(¶ms))?;
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 => {
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 {
order_prev + 1
} else {
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(¶ms))?;
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 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(¶ms))?;
}
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(¶ms))?.len() < 1 {
tracing::warn!("Now row with _id {id} found to move");
new_order = 0;
}
Ok(new_order)
}
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(¶ms))?
.ok_or(RelatableError::DataError(format!("No row with _id = {id}")))?
.get_unsigned("_id")?;
Ok(())
}
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)
}
pub async fn validate_table(&self, table: &Table) -> Result<()> {
tracing::trace!("Relatable::validate_table({self:?}, {table:?})");
let mut conn = self.connection.reconnect()?;
let mut tx = self.connection.begin(&mut conn).await?;
self._validate_table(table, &mut tx)?;
tx.commit()?;
tracing::info!("Validated table '{}'", table.name);
Ok(())
}
fn _validate_table(&self, table: &Table, tx: &mut DbTransaction<'_>) -> Result<()> {
tracing::trace!("Relatable::_validate_table({self:?}, {table:?}, tx)");
for (_, column) in table.columns.iter() {
self._validate_column_optionally_for_row(column, None, tx)?;
}
tracing::debug!("Validated table '{}'", table.name);
Ok(())
}
pub async fn validate_datatype_for_table(&self, table: &Table) -> Result<()> {
tracing::trace!("Relatable::validate_datatype_for_table({self:?}, {table:?})");
let mut conn = self.connection.reconnect()?;
let mut tx = self.connection.begin(&mut conn).await?;
self._validate_datatype_for_table(table, &mut tx)?;
tx.commit()?;
tracing::info!("Validated datatype for table '{}'", table.name);
Ok(())
}
fn _validate_datatype_for_table(
&self,
table: &Table,
tx: &mut DbTransaction<'_>,
) -> Result<()> {
tracing::trace!("Relatable::_validate_datatype_for_table({self:?}, {table:?}, tx)");
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(())
}
pub async fn validate_structure_for_table(&self, table: &Table) -> Result<()> {
tracing::trace!("Relatable::validate_structure_for_table({self:?}, {table:?})");
let mut conn = self.connection.reconnect()?;
let mut tx = self.connection.begin(&mut conn).await?;
self._validate_structure_for_table(table, &mut tx)?;
tx.commit()?;
tracing::info!("Validated structure for table '{}'", table.name);
Ok(())
}
fn _validate_structure_for_table(
&self,
table: &Table,
tx: &mut DbTransaction<'_>,
) -> Result<()> {
tracing::trace!("Relatable::_validate_structure_for_table({self:?}, {table:?}, tx)");
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(())
}
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(())
}
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(())
}
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(())
}
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(())
}
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();
self._delete_message(
tx,
table_name,
row.copied(),
Some(&column.name),
Some("datatype:%"),
Some("rltbl"),
)?;
let mut datatypes_to_check = vec![column.datatype.clone()];
datatypes_to_check.append(&mut column.datatype_hierarchy.clone());
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(())
}
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();
self._delete_message(
tx,
table_name,
row.copied(),
Some(&column.name),
Some("key:%"),
Some("rltbl"),
)?;
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(())
}
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(())
}
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 => {
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(¶ms))?;
} else {
tracing::debug!("Truncating cache");
tx.query(&sql, None)?;
}
Ok(())
}
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);
}
}
}
}
#[derive(Clone, Copy, Debug, PartialEq, Serialize, Deserialize)]
pub enum ValidationLevel {
None,
SqlType,
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(),
);
}
}
}
}
#[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 {
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()),
}
}
}
#[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"),
}
}
}
#[derive(Clone, Debug, Serialize, Deserialize)]
#[serde(tag = "type")]
pub enum Change {
Update {
row: u64,
column: String,
before: JsonValue,
after: JsonValue,
},
Add {
row: u64,
after: u64,
},
Move {
row: u64,
from_after: u64,
to_after: u64,
},
Delete {
row: u64,
after: u64,
},
}
impl Change {
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)
}
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());
}
}
}
}
#[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}"),
}
}
}
#[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,
pub columns: Vec<Column>,
pub rows: Vec<Row>,
}
impl ResultSet {
pub fn to_csv(&self) -> String {
let writer = WriterBuilder::new().from_writer(vec![]);
self.to_xsv(writer)
}
pub fn to_tsv(&self) -> String {
let writer = WriterBuilder::new()
.delimiter(b'\t')
.quote_style(QuoteStyle::Never)
.from_writer(vec![]);
self.to_xsv(writer)
}
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()
}
fn format_cell_text_value(column_format: &str, format_regex: &Regex, cell: &str) -> String {
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) => {
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()
}
}
}
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}")
}
}
#[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,
}