filipriec/komp_ac
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fixing post with links

Browse Source
This commit is contained in:
filipriec
2025-06-16 14:42:49 +02:00
parent 9e0fa9ddb1
commit c31f08d5b8
11 changed files with 169 additions and 121 deletions
Download Patch File Download Diff File Expand all files Collapse all files

108
server/src/tables_data/handlers/post_table_data.rs
View File

@@ -8,16 +8,15 @@ use common::proto::multieko2::tables_data::{PostTableDataRequest, PostTableDataR
use std::collections::HashMap;
use std::sync::Arc;
use crate::shared::schema_qualifier::qualify_table_name_for_data;
use prost_types::value::Kind; // NEW: Import the Kind enum
use crate::steel::server::execution::{self, Value};
use crate::steel::server::functions::SteelContext;
// Add these imports
use crate::indexer::{IndexCommand, IndexCommandData};
use tokio::sync::mpsc;
use tracing::error;
// MODIFIED: Function signature now accepts the indexer sender
pub async fn post_table_data(
db_pool: &PgPool,
request: PostTableDataRequest,
@@ -25,11 +24,12 @@ pub async fn post_table_data(
) -> Result<PostTableDataResponse, Status> {
let profile_name = request.profile_name;
let table_name = request.table_name;
let mut data = HashMap::new();
for (key, value) in request.data {
data.insert(key, value.trim().to_string());
}
// REMOVED: The old data conversion loop. We will process request.data directly.
// let mut data = HashMap::new();
// for (key, value) in request.data {
// data.insert(key, value.trim().to_string());
// }
// Lookup profile
let profile = sqlx::query!(
@@ -94,13 +94,28 @@ pub async fn post_table_data(
// Validate all data columns
let user_columns: Vec<&String> = columns.iter().map(|(name, _)| name).collect();
for key in data.keys() {
for key in request.data.keys() { // CHANGED: Use request.data
if !system_columns_set.contains(key.as_str()) &&
!user_columns.contains(&&key.to_string()) {
return Err(Status::invalid_argument(format!("Invalid column: {}", key)));
}
}
// NEW: Create a string-based map for backwards compatibility with Steel scripts.
let mut string_data_for_scripts = HashMap::new();
for (key, proto_value) in &request.data {
let str_val = match &proto_value.kind {
Some(Kind::StringValue(s)) => s.clone(),
Some(Kind::NumberValue(n)) => n.to_string(),
Some(Kind::BoolValue(b)) => b.to_string(),
Some(Kind::NullValue(_)) => String::new(),
Some(Kind::StructValue(_)) | Some(Kind::ListValue(_)) | None => {
return Err(Status::invalid_argument(format!("Unsupported type for script validation in column '{}'", key)));
}
};
string_data_for_scripts.insert(key.clone(), str_val);
}
// Validate Steel scripts
let scripts = sqlx::query!(
"SELECT target_column, script FROM table_scripts WHERE table_definitions_id = $1",
@@ -114,20 +129,20 @@ pub async fn post_table_data(
let target_column = script_record.target_column;
// Ensure target column exists in submitted data
let user_value = data.get(&target_column)
let user_value = string_data_for_scripts.get(&target_column) // CHANGED: Use the new string map
.ok_or_else(|| Status::invalid_argument(
format!("Script target column '{}' is required", target_column)
))?;
// Create execution context
let context = SteelContext {
current_table: table_name.clone(), // Keep base name for scripts
current_table: table_name.clone(),
profile_id,
row_data: data.clone(),
row_data: string_data_for_scripts.clone(), // CHANGED: Use the new string map
db_pool: Arc::new(db_pool.clone()),
};
// Execute validation script
// ... (rest of script execution is the same)
let script_result = execution::execute_script(
script_record.script,
"STRINGS",
@@ -138,7 +153,6 @@ pub async fn post_table_data(
format!("Script execution failed for '{}': {}", target_column, e)
))?;
// Validate script output
let Value::Strings(mut script_output) = script_result else {
return Err(Status::internal("Script must return string values"));
};
@@ -160,11 +174,12 @@ pub async fn post_table_data(
let mut placeholders = Vec::new();
let mut param_idx = 1;
for (col, value) in data {
// CHANGED: This entire loop is rewritten to handle prost_types::Value
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", // Handle foreign keys
_ if col.ends_with("_id") => "BIGINT",
_ => return Err(Status::invalid_argument("Invalid system column")),
}
} else {
@@ -174,36 +189,55 @@ pub async fn post_table_data(
.ok_or_else(|| Status::invalid_argument(format!("Column not found: {}", col)))?
};
let kind = proto_value.kind.ok_or_else(|| {
Status::invalid_argument(format!("Value for column '{}' cannot be null", col))
})?;
match sql_type {
"TEXT" | "VARCHAR(15)" | "VARCHAR(255)" => {
if let Some(max_len) = sql_type.strip_prefix("VARCHAR(")
.and_then(|s| s.strip_suffix(')'))
.and_then(|s| s.parse::<usize>().ok())
{
if value.len() > max_len {
return Err(Status::internal(format!("Value too long for {}", col)));
if let Kind::StringValue(value) = kind {
if let Some(max_len) = sql_type.strip_prefix("VARCHAR(")
.and_then(|s| s.strip_suffix(')'))
.and_then(|s| s.parse::<usize>().ok())
{
if value.len() > max_len {
return Err(Status::internal(format!("Value too long for {}", col)));
}
}
params.add(value)
.map_err(|e| Status::invalid_argument(format!("Failed to add text parameter for {}: {}", col, e)))?;
} else {
return Err(Status::invalid_argument(format!("Expected string for column '{}'", col)));
}
params.add(value)
.map_err(|e| Status::invalid_argument(format!("Failed to add text parameter for {}: {}", col, e)))?;
},
"BOOLEAN" => {
let val = value.parse::<bool>()
.map_err(|_| Status::invalid_argument(format!("Invalid boolean for {}", col)))?;
params.add(val)
.map_err(|e| Status::invalid_argument(format!("Failed to add boolean parameter for {}: {}", col, e)))?;
if let Kind::BoolValue(val) = kind {
params.add(val)
.map_err(|e| Status::invalid_argument(format!("Failed to add boolean parameter for {}: {}", col, e)))?;
} else {
return Err(Status::invalid_argument(format!("Expected boolean for column '{}'", col)));
}
},
"TIMESTAMPTZ" => {
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::invalid_argument(format!("Failed to add timestamp parameter for {}: {}", col, e)))?;
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::invalid_argument(format!("Failed to add timestamp parameter for {}: {}", col, e)))?;
} else {
return Err(Status::invalid_argument(format!("Expected ISO 8601 string for column '{}'", col)));
}
},
"BIGINT" => {
let val = value.parse::<i64>()
.map_err(|_| Status::invalid_argument(format!("Invalid integer for {}", col)))?;
params.add(val)
.map_err(|e| Status::invalid_argument(format!("Failed to add integer parameter for {}: {}", col, e)))?;
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::invalid_argument(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))),
}
@@ -227,7 +261,7 @@ pub async fn post_table_data(
placeholders.join(", ")
);
// Execute query with enhanced error handling
// ... (rest of the function is unchanged)
let result = sqlx::query_scalar_with::<_, i64, _>(&sql, params)
.fetch_one(db_pool)
.await;
@@ -235,7 +269,6 @@ pub async fn post_table_data(
let inserted_id = match result {
Ok(id) => id,
Err(e) => {
// Handle "relation does not exist" error specifically
if let Some(db_err) = e.as_database_error() {
if db_err.code() == Some(std::borrow::Cow::Borrowed("42P01")) {
return Err(Status::internal(format!(
@@ -248,15 +281,12 @@ pub async fn post_table_data(
}
};
// After a successful insert, send a command to the indexer.
let command = IndexCommand::AddOrUpdate(IndexCommandData {
table_name: table_name.clone(),
row_id: inserted_id,
});
if let Err(e) = indexer_tx.send(command).await {
// If sending fails, the DB is updated but the index will be stale.
// This is a critical situation to log and monitor.
error!(
"CRITICAL: DB insert for table '{}' (id: {}) succeeded but failed to queue for indexing: {}. Search index is now inconsistent.",
table_name, inserted_id, e