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

table name

Browse Source
This commit is contained in:
Filipriec
2026-08-25 13:01:32 +02:00
parent ed01d1ec5b
commit aea54f8d33
12 changed files with 135 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

99
search/src/lib.rs
View File

@@ -165,7 +165,7 @@ impl SearcherService {
request: Request<SearchRequest>,
) -> Result<Response<SearchResponse>, Status> {
let req = request.into_inner();
let normalized = normalize_request(req)?;
let mut normalized = normalize_request(req)?;
if !profile_exists(&self.pool, &normalized.profile_name).await? {
return Err(Status::not_found(format!(
@@ -174,13 +174,15 @@ impl SearcherService {
)));
}
if let Some(table_name) = normalized.table_name.as_deref() {
if !table_exists(&self.pool, &normalized.profile_name, table_name).await? {
return Err(Status::not_found(format!(
"Table '{}' was not found in profile '{}'",
table_name, normalized.profile_name
)));
}
if let Some(table_name) = normalized.table_name.clone() {
normalized.table_name = Some(
visible_table_name(&self.pool, &normalized.profile_name, &table_name)
.await?
.ok_or_else(|| Status::not_found(format!(
"Table '{}' was not found in profile '{}'",
table_name, normalized.profile_name
)))?,
);
}
if !normalized.has_input() {
@@ -266,8 +268,8 @@ impl SearcherService {
&self,
request: Request<SearchRequest>,
) -> Result<Response<SearchCountResponse>, Status> {
let normalized = normalize_request(request.into_inner())?;
let table_name = normalized.table_name.as_deref().ok_or_else(|| {
let mut normalized = normalize_request(request.into_inner())?;
let requested_table_name = normalized.table_name.clone().ok_or_else(|| {
Status::invalid_argument("table_name is required when counting search results")
})?;
if !normalized.has_input() {
@@ -275,11 +277,17 @@ impl SearcherService {
"counting search results requires text or a column constraint",
));
}
if !profile_exists(&self.pool, &normalized.profile_name).await?
|| !table_exists(&self.pool, &normalized.profile_name, table_name).await?
{
if !profile_exists(&self.pool, &normalized.profile_name).await? {
return Err(Status::not_found("Search table was not found"));
}
normalized.table_name = visible_table_name(
&self.pool,
&normalized.profile_name,
&requested_table_name,
)
.await?;
let table_name = normalized.table_name.as_deref()
.ok_or_else(|| Status::not_found("Search table was not found"))?;
let index_path = search_index_path(
&common::search::search_index_root(),
&normalized.profile_name,
@@ -744,6 +752,25 @@ fn validate_identifier(value: &str, field_name: &str) -> Result<(), Status> {
Ok(())
}
fn validate_table_name(value: &str) -> Result<(), Status> {
if value.is_empty()
|| !value.chars().all(|character| {
character.is_ascii_alphanumeric() || matches!(character, '_' | '-')
})
|| !value
.chars()
.next()
.is_some_and(|character| character.is_ascii_alphanumeric())
|| !value
.chars()
.last()
.is_some_and(|character| character.is_ascii_alphanumeric())
{
return Err(Status::invalid_argument("table_name contains invalid characters"));
}
Ok(())
}
fn validate_search_column(value: &str) -> Result<(), Status> {
if value.is_empty() {
return Err(Status::invalid_argument(
@@ -774,25 +801,26 @@ async fn profile_exists(pool: &PgPool, profile_name: &str) -> Result<bool, Statu
Ok(exists)
}
async fn table_exists(pool: &PgPool, profile_name: &str, table_name: &str) -> Result<bool, Status> {
let exists = sqlx::query_scalar::<_, bool>(
r#"
SELECT EXISTS(
SELECT 1
FROM table_definitions td
JOIN schemas s ON td.schema_id = s.id
WHERE td.table_name = $2
AND td.deleted = FALSE
AND (s.name = $1 OR td.is_global = TRUE)
)
"#,
async fn visible_table_name(
pool: &PgPool,
profile_name: &str,
table_name: &str,
) -> Result<Option<String>, Status> {
sqlx::query_scalar::<_, String>(
r#"SELECT td.table_name
FROM table_definitions td
JOIN schemas s ON td.schema_id = s.id
WHERE td.canonical_table_name = $2
AND td.deleted = FALSE
AND (s.name = $1 OR td.is_global = TRUE)
ORDER BY td.is_global ASC
LIMIT 1"#,
)
.bind(profile_name)
.bind(table_name)
.fetch_one(pool)
.bind(common::alias::canonical_table_name(table_name))
.fetch_optional(pool)
.await
.map_err(|e| Status::internal(format!("Table lookup failed: {}", e)))?;
Ok(exists)
.map_err(|e| Status::internal(format!("Table lookup failed: {}", e)))
}
async fn qualified_visible_table(
@@ -800,23 +828,24 @@ async fn qualified_visible_table(
profile_name: &str,
table_name: &str,
) -> Result<String, Status> {
let storage_schema = sqlx::query_scalar::<_, String>(
r#"SELECT owner.name
let resolved = sqlx::query_as::<_, (String, String)>(
r#"SELECT owner.name, definition.table_name
FROM table_definitions definition
JOIN schemas owner ON owner.id = definition.schema_id
WHERE definition.table_name = $2
WHERE definition.canonical_table_name = $2
AND definition.deleted = FALSE
AND (owner.name = $1 OR definition.is_global = TRUE)
ORDER BY definition.is_global ASC
LIMIT 1"#,
)
.bind(profile_name)
.bind(table_name)
.bind(common::alias::canonical_table_name(table_name))
.fetch_optional(pool)
.await
.map_err(|error| Status::internal(format!("Table storage lookup failed: {error}")))?
.ok_or_else(|| Status::not_found(format!("Table '{table_name}' was not found")))?;
Ok(qualify_profile_table(&storage_schema, table_name))
let (storage_schema, stored_table_name) = resolved;
Ok(qualify_profile_table(&storage_schema, &stored_table_name))
}
fn normalize_request(req: SearchRequest) -> Result<NormalizedSearchRequest, Status> {
@@ -828,7 +857,7 @@ fn normalize_request(req: SearchRequest) -> Result<NormalizedSearchRequest, Stat
let table_name = match req.table_name.as_deref().map(str::trim) {
Some(table_name) if !table_name.is_empty() => {
validate_identifier(table_name, "table_name")?;
validate_table_name(table_name)?;
Some(table_name.to_string())
}
_ => None,