197 lines
7.9 KiB
Rust
197 lines
7.9 KiB
Rust
// src/tables_data/handlers/put_table_data.rs
|
|
use tonic::Status;
|
|
use sqlx::{PgPool, Arguments, Postgres};
|
|
use sqlx::postgres::PgArguments;
|
|
use chrono::{DateTime, Utc};
|
|
use common::proto::multieko2::tables_data::{PutTableDataRequest, PutTableDataResponse};
|
|
use crate::shared::schema_qualifier::qualify_table_name_for_data;
|
|
use prost_types::value::Kind;
|
|
|
|
pub async fn put_table_data(
|
|
db_pool: &PgPool,
|
|
request: PutTableDataRequest,
|
|
) -> Result<PutTableDataResponse, Status> {
|
|
let profile_name = request.profile_name;
|
|
let table_name = request.table_name;
|
|
let record_id = request.id;
|
|
|
|
// If no data is provided to update, it's an invalid request.
|
|
if request.data.is_empty() {
|
|
return Err(Status::invalid_argument("No fields provided to update."));
|
|
}
|
|
|
|
// Lookup profile
|
|
let profile = sqlx::query!(
|
|
"SELECT id FROM profiles WHERE name = $1",
|
|
profile_name
|
|
)
|
|
.fetch_optional(db_pool)
|
|
.await
|
|
.map_err(|e| Status::internal(format!("Profile lookup error: {}", e)))?;
|
|
|
|
let profile_id = profile.ok_or_else(|| Status::not_found("Profile not found"))?.id;
|
|
|
|
// Lookup table_definition
|
|
let table_def = sqlx::query!(
|
|
r#"SELECT id, columns FROM table_definitions
|
|
WHERE profile_id = $1 AND table_name = $2"#,
|
|
profile_id,
|
|
table_name
|
|
)
|
|
.fetch_optional(db_pool)
|
|
.await
|
|
.map_err(|e| Status::internal(format!("Table lookup error: {}", e)))?;
|
|
|
|
let table_def = table_def.ok_or_else(|| Status::not_found("Table not found"))?;
|
|
|
|
// Parse columns from JSON
|
|
let columns_json: Vec<String> = serde_json::from_value(table_def.columns.clone())
|
|
.map_err(|e| Status::internal(format!("Column parsing error: {}", e)))?;
|
|
|
|
let mut columns = Vec::new();
|
|
for col_def in columns_json {
|
|
let parts: Vec<&str> = col_def.splitn(2, ' ').collect();
|
|
if parts.len() != 2 {
|
|
return Err(Status::internal("Invalid column format"));
|
|
}
|
|
let name = parts[0].trim_matches('"').to_string();
|
|
let sql_type = parts[1].to_string();
|
|
columns.push((name, sql_type));
|
|
}
|
|
|
|
// Get all foreign key columns for this table (needed for validation)
|
|
let fk_columns = sqlx::query!(
|
|
r#"SELECT ltd.table_name
|
|
FROM table_definition_links tdl
|
|
JOIN table_definitions ltd ON tdl.linked_table_id = ltd.id
|
|
WHERE tdl.source_table_id = $1"#,
|
|
table_def.id
|
|
)
|
|
.fetch_all(db_pool)
|
|
.await
|
|
.map_err(|e| Status::internal(format!("Foreign key lookup error: {}", e)))?;
|
|
|
|
let mut system_columns = vec!["deleted".to_string()];
|
|
for fk in fk_columns {
|
|
let base_name = fk.table_name.split_once('_').map_or(fk.table_name.as_str(), |(_, rest)| rest);
|
|
system_columns.push(format!("{}_id", base_name));
|
|
}
|
|
let system_columns_set: std::collections::HashSet<_> = system_columns.iter().map(|s| s.as_str()).collect();
|
|
let user_columns: Vec<&String> = columns.iter().map(|(name, _)| name).collect();
|
|
|
|
// Validate input columns
|
|
for key in request.data.keys() {
|
|
if !system_columns_set.contains(key.as_str()) && !user_columns.contains(&key) {
|
|
return Err(Status::invalid_argument(format!("Invalid column: {}", key)));
|
|
}
|
|
}
|
|
|
|
// Prepare SQL parameters
|
|
let mut params = PgArguments::default();
|
|
let mut set_clauses = Vec::new();
|
|
let mut param_idx = 1;
|
|
|
|
for (col, proto_value) in request.data {
|
|
let sql_type = if system_columns_set.contains(col.as_str()) {
|
|
match col.as_str() {
|
|
"deleted" => "BOOLEAN",
|
|
_ if col.ends_with("_id") => "BIGINT",
|
|
_ => return Err(Status::invalid_argument("Invalid system column")),
|
|
}
|
|
} else {
|
|
columns.iter()
|
|
.find(|(name, _)| name == &col)
|
|
.map(|(_, sql_type)| sql_type.as_str())
|
|
.ok_or_else(|| Status::invalid_argument(format!("Column not found: {}", col)))?
|
|
};
|
|
|
|
// A provided value cannot be null or empty in a PUT request.
|
|
// To clear a field, it should be set to an empty string "" for text,
|
|
// or a specific value for other types if needed (though typically not done).
|
|
// For now, we reject nulls.
|
|
let kind = proto_value.kind.ok_or_else(|| {
|
|
Status::invalid_argument(format!("Value for column '{}' cannot be empty in a PUT request. To clear a text field, send an empty string.", col))
|
|
})?;
|
|
|
|
match sql_type {
|
|
"TEXT" | "VARCHAR(15)" | "VARCHAR(255)" => {
|
|
if let Kind::StringValue(value) = kind {
|
|
params.add(value)
|
|
.map_err(|e| Status::internal(format!("Failed to add text parameter for {}: {}", col, e)))?;
|
|
} else {
|
|
return Err(Status::invalid_argument(format!("Expected string for column '{}'", col)));
|
|
}
|
|
},
|
|
"BOOLEAN" => {
|
|
if let Kind::BoolValue(val) = kind {
|
|
params.add(val)
|
|
.map_err(|e| Status::internal(format!("Failed to add boolean parameter for {}: {}", col, e)))?;
|
|
} else {
|
|
return Err(Status::invalid_argument(format!("Expected boolean for column '{}'", col)));
|
|
}
|
|
},
|
|
"TIMESTAMPTZ" => {
|
|
if let Kind::StringValue(value) = kind {
|
|
let dt = DateTime::parse_from_rfc3339(&value)
|
|
.map_err(|_| Status::invalid_argument(format!("Invalid timestamp for {}", col)))?;
|
|
params.add(dt.with_timezone(&Utc))
|
|
.map_err(|e| Status::internal(format!("Failed to add timestamp parameter for {}: {}", col, e)))?;
|
|
} else {
|
|
return Err(Status::invalid_argument(format!("Expected ISO 8601 string for column '{}'", col)));
|
|
}
|
|
},
|
|
"BIGINT" => {
|
|
if let Kind::NumberValue(val) = kind {
|
|
if val.fract() != 0.0 {
|
|
return Err(Status::invalid_argument(format!("Expected integer for column '{}', but got a float", col)));
|
|
}
|
|
params.add(val as i64)
|
|
.map_err(|e| Status::internal(format!("Failed to add integer parameter for {}: {}", col, e)))?;
|
|
} else {
|
|
return Err(Status::invalid_argument(format!("Expected number for column '{}'", col)));
|
|
}
|
|
},
|
|
_ => return Err(Status::invalid_argument(format!("Unsupported type {}", sql_type))),
|
|
}
|
|
|
|
set_clauses.push(format!("\"{}\" = ${}", col, param_idx));
|
|
param_idx += 1;
|
|
}
|
|
|
|
params.add(record_id)
|
|
.map_err(|e| Status::internal(format!("Failed to add record_id parameter: {}", e)))?;
|
|
|
|
let qualified_table = qualify_table_name_for_data(&table_name)?;
|
|
let set_clause = set_clauses.join(", ");
|
|
let sql = format!(
|
|
"UPDATE {} SET {} WHERE id = ${} AND deleted = FALSE RETURNING id",
|
|
qualified_table,
|
|
set_clause,
|
|
param_idx
|
|
);
|
|
|
|
let result = sqlx::query_scalar_with::<Postgres, i64, _>(&sql, params)
|
|
.fetch_optional(db_pool)
|
|
.await;
|
|
|
|
match result {
|
|
Ok(Some(updated_id)) => Ok(PutTableDataResponse {
|
|
success: true,
|
|
message: "Data updated successfully".into(),
|
|
updated_id,
|
|
}),
|
|
Ok(None) => Err(Status::not_found("Record not found or already deleted")),
|
|
Err(e) => {
|
|
if let Some(db_err) = e.as_database_error() {
|
|
if db_err.code() == Some(std::borrow::Cow::Borrowed("42P01")) {
|
|
return Err(Status::internal(format!(
|
|
"Table '{}' is defined but does not physically exist in the database as {}",
|
|
table_name, qualified_table
|
|
)));
|
|
}
|
|
}
|
|
Err(Status::internal(format!("Update failed: {}", e)))
|
|
}
|
|
}
|
|
}
|