mod query_builder; use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet}; use std::path::Path; use std::sync::{Arc, Mutex}; use common::proto::komp_ac::search::searcher_server::Searcher; pub use common::proto::komp_ac::search::searcher_server::SearcherServer; use common::proto::komp_ac::search::{ SearchCountResponse, SearchOrderDirection, SearchRequest, SearchResponse, SearchVersionScope, search_response::Hit, }; use common::search::{ F_IS_ARCHIVED, F_PG_ID, F_ROW_REVISION, F_TABLE_DEFINITION_ID, F_TABLE_NAME, F_VERSION_NUMBER, SchemaFields, canonical_exact_search_value, parse_archived_search_row_key, register_tokenizers, search_index_path, search_row_key, }; use common::system_column::{internal_column_names, is_internal_column, is_system_column}; use query_builder::{ ConstraintMode, SearchConstraint, SearchConstraintTarget, build_master_query, }; use sqlx::{AssertSqlSafe, PgPool, Row}; use tantivy::collector::TopDocs; use tantivy::query::Query; use tantivy::{Index, IndexReader, ReloadPolicy}; use tonic::{Request, Response, Status}; use tracing::info; const DEFAULT_RESULT_LIMIT: usize = 60; const HARD_RESULT_LIMIT: usize = 200; const DEFAULT_LIST_LIMIT: usize = 5; const SEARCH_SCORE_RELATIVE_FLOOR: f32 = 0.25; const SEARCH_SCORE_GROUP_WIDTH: f32 = 0.25; const LEDGER_ACCOUNTS_TABLE_NAME: &str = "ledger_accounts"; pub struct AccountPathProjection { paths: HashMap, account_column: Option, accounts_table: bool, } impl AccountPathProjection { pub async fn load( pool: &PgPool, profile_name: &str, table_name: &str, physical_to_display: &HashMap, ) -> Result { let account_column = physical_to_display .get(common::system_column::ACCOUNT_REFERENCE_COLUMN) .cloned(); let accounts_table = table_name == LEDGER_ACCOUNTS_TABLE_NAME; if account_column.is_none() && !accounts_table { return Ok(Self { paths: HashMap::new(), account_column, accounts_table, }); } let qualified_accounts = format!( "\"{}\".\"{}\"", profile_name.replace('"', "\"\""), LEDGER_ACCOUNTS_TABLE_NAME, ); let rows = sqlx::query(AssertSqlSafe(format!( "SELECT id, segment, parent_account_id FROM {qualified_accounts} WHERE deleted = FALSE" ))) .fetch_all(pool) .await .map_err(|error| Status::internal(format!("Account path lookup failed: {error}")))?; let mut nodes = BTreeMap::new(); for row in rows { let id: i64 = row.try_get("id").map_err(|error| { Status::internal(format!("Account id read failed: {error}")) })?; let segment: String = row.try_get("segment").map_err(|error| { Status::internal(format!("Account segment read failed: {error}")) })?; let parent: Option = row.try_get("parent_account_id").map_err(|error| { Status::internal(format!("Account parent read failed: {error}")) })?; nodes.insert(id, (segment, parent)); } let mut paths = HashMap::new(); for &id in nodes.keys() { let mut segments = Vec::new(); let mut visited = BTreeSet::new(); let mut current = Some(id); while let Some(node_id) = current { if !visited.insert(node_id) { return Err(Status::internal("Stored account hierarchy contains a cycle")); } let (segment, parent) = nodes.get(&node_id).ok_or_else(|| { Status::internal(format!("Account {node_id} was not found")) })?; segments.push(segment.clone()); current = *parent; } segments.reverse(); paths.insert(id, segments.join("/")); } Ok(Self { paths, account_column, accounts_table, }) } pub fn apply(&self, row_id: i64, value: &mut serde_json::Value) -> Result<(), Status> { let Some(object) = value.as_object_mut() else { return Ok(()); }; if let Some(column) = &self.account_column { if let Some(stored) = object.get(column).filter(|value| !value.is_null()) { let account_id = match stored { serde_json::Value::Number(value) => value.as_i64(), serde_json::Value::String(value) => value.parse().ok(), _ => None, } .ok_or_else(|| Status::internal("Stored account reference is invalid"))?; let path = self.paths.get(&account_id).ok_or_else(|| { Status::internal(format!("Account {account_id} was not found")) })?; object.insert(column.clone(), serde_json::Value::String(path.clone())); } } if self.accounts_table { let path = self.paths.get(&row_id).ok_or_else(|| { Status::internal(format!("Account {row_id} was not found")) })?; object.insert( common::system_column::ACCOUNT_API_COLUMN.to_string(), serde_json::Value::String(path.clone()), ); } Ok(()) } pub fn row_display_columns(&self, columns: &[String]) -> Vec { if self.accounts_table { vec![common::system_column::ACCOUNT_API_COLUMN.to_string()] } else { columns.to_vec() } } } pub struct SearcherService { pub pool: PgPool, profiles: Mutex>>, } impl SearcherService { pub fn new(pool: PgPool) -> Self { Self { pool, profiles: Mutex::new(HashMap::new()), } } async fn run_rpc( &self, request: Request, ) -> Result, Status> { let req = request.into_inner(); let normalized = normalize_request(req)?; if !profile_exists(&self.pool, &normalized.profile_name).await? { return Err(Status::not_found(format!( "Profile '{}' was not found", normalized.profile_name ))); } 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 !normalized.has_input() { if normalized.version_scope != SearchVersionScope::Current { return Err(Status::invalid_argument( "archived searches require text or a column constraint", )); } let Some(table_name) = normalized.table_name.as_deref() else { return Err(Status::invalid_argument( "table_name is required when query is empty", )); }; let hits = fetch_ordered_rows( &self.pool, &normalized.profile_name, table_name, normalized.order.as_ref(), normalized.limit.unwrap_or(DEFAULT_LIST_LIMIT), normalized.offset, ) .await?; return Ok(Response::new(SearchResponse { hits })); } let index_path = search_index_path( &common::search::search_index_root(), &normalized.profile_name, ); if !index_path.exists() { return Err(Status::not_found(format!( "No search index found for profile '{}'", normalized.profile_name ))); } let resolved_must = resolve_constraints( &self.pool, &normalized.profile_name, normalized.table_name.as_deref(), &normalized.must, ) .await?; let profile = profile_index(&self.profiles, &normalized.profile_name, &index_path)?; let mut hits = run_search( &self.pool, &profile, &normalized.profile_name, normalized.table_name.as_deref(), &normalized.free_query, &resolved_must, normalized.limit.unwrap_or(DEFAULT_RESULT_LIMIT), normalized.offset, normalized.order.as_ref(), normalized.version_scope, ) .await?; if normalized.order.is_none() { hits.sort_by(|left, right| right.score.total_cmp(&left.score)); if let Some(limit) = normalized.limit { if hits.len() > limit { hits.truncate(limit); } } } info!( "search: profile={} table={:?} free='{}' constraints={} hits={}", normalized.profile_name, normalized.table_name, normalized.free_query, resolved_must.len(), hits.len() ); Ok(Response::new(SearchResponse { hits })) } async fn run_count_rpc( &self, request: Request, ) -> Result, Status> { let normalized = normalize_request(request.into_inner())?; let table_name = normalized.table_name.as_deref().ok_or_else(|| { Status::invalid_argument("table_name is required when counting search results") })?; if !normalized.has_input() { return Err(Status::invalid_argument( "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? { return Err(Status::not_found("Search table was not found")); } let index_path = search_index_path( &common::search::search_index_root(), &normalized.profile_name, ); if !index_path.exists() { return Err(Status::not_found(format!( "No search index found for profile '{}'", normalized.profile_name ))); } let constraints = resolve_constraints( &self.pool, &normalized.profile_name, Some(table_name), &normalized.must, ) .await?; let profile = profile_index(&self.profiles, &normalized.profile_name, &index_path)?; let query = build_master_query( &profile.index, &profile.fields, &normalized.free_query, &constraints, Some(table_name), normalized.version_scope, )?; let count = count_authoritative_matches( &self.pool, &profile, &normalized.profile_name, table_name, &*query, ) .await?; Ok(Response::new(SearchCountResponse { count })) } } async fn count_authoritative_matches( pool: &PgPool, profile: &ProfileIndex, profile_name: &str, table_name: &str, query: &dyn Query, ) -> Result { let searcher = profile.reader.searcher(); let documents = top_documents_for_count(&searcher, query)?; let best_score = documents .first() .map(|(score, _)| *score) .unwrap_or_default(); let score_floor = if best_score > 0.0 { best_score * SEARCH_SCORE_RELATIVE_FLOOR } else { 0.0 }; let mut documents_by_segment: HashMap> = HashMap::new(); for (score, address) in documents { if score >= score_floor { documents_by_segment .entry(address.segment_ord) .or_default() .push(address.doc_id); } } let mut current_rows = Vec::new(); let mut archives = Vec::new(); for (segment_ord, segment_documents) in documents_by_segment { let segment = searcher.segment_reader(segment_ord); let fast_fields = segment.fast_fields(); let pg_ids = fast_fields.u64(F_PG_ID).map_err(|error| { Status::internal(format!("Search count fast field '{F_PG_ID}' failed: {error}")) })?; let archived_flags = fast_fields.u64(F_IS_ARCHIVED).map_err(|error| { Status::internal(format!( "Search count fast field '{F_IS_ARCHIVED}' failed: {error}" )) })?; let revisions = fast_fields.u64(F_ROW_REVISION).map_err(|error| { Status::internal(format!( "Search count fast field '{F_ROW_REVISION}' failed: {error}" )) })?; let definition_ids = fast_fields.u64(F_TABLE_DEFINITION_ID).map_err(|error| { Status::internal(format!( "Search count fast field '{F_TABLE_DEFINITION_ID}' failed: {error}" )) })?; let versions = fast_fields.u64(F_VERSION_NUMBER).map_err(|error| { Status::internal(format!( "Search count fast field '{F_VERSION_NUMBER}' failed: {error}" )) })?; let read = |column: &tantivy::columnar::Column, doc_id, field_name: &str| -> Result { column .first(doc_id) .and_then(|value| i64::try_from(value).ok()) .ok_or_else(|| { Status::internal(format!( "Search count document has no valid '{field_name}'" )) }) }; for doc_id in segment_documents { let row_id = read(&pg_ids, doc_id, F_PG_ID)?; if read(&archived_flags, doc_id, F_IS_ARCHIVED)? != 0 { archives.push(( read(&definition_ids, doc_id, F_TABLE_DEFINITION_ID)?, row_id, read(&versions, doc_id, F_VERSION_NUMBER)?, )); } else { current_rows.push((row_id, read(&revisions, doc_id, F_ROW_REVISION)?)); } } } current_rows.sort_unstable(); current_rows.dedup(); archives.sort_unstable(); archives.dedup(); count_validated_candidates( pool, profile_name, table_name, ¤t_rows, &archives, ) .await } fn top_documents_for_count( searcher: &tantivy::Searcher, query: &dyn Query, ) -> Result, Status> { if searcher.num_docs() == 0 { return Ok(Vec::new()); } // DocAddress is local to this exact Searcher snapshot. The caller must use // the same searcher to load every address returned here. searcher .search( query, &TopDocs::with_limit(searcher.num_docs() as usize).order_by_score(), ) .map_err(|error| Status::internal(format!("Search count failed: {error}"))) } async fn count_validated_candidates( pool: &PgPool, profile_name: &str, table_name: &str, current_rows: &[(i64, i64)], archives: &[(i64, i64, i64)], ) -> Result { let current_count = if current_rows.is_empty() { 0 } else { let qualified_table = qualified_visible_table(pool, profile_name, table_name).await?; let current_ids = current_rows.iter().map(|item| item.0).collect::>(); let current_revisions = current_rows.iter().map(|item| item.1).collect::>(); sqlx::query_scalar::<_, i64>(AssertSqlSafe(format!( "SELECT COUNT(*) FROM {qualified_table} current_row \ JOIN UNNEST($1::BIGINT[], $2::BIGINT[]) \ AS candidate(id, row_revision) \ ON candidate.id = current_row.id \ AND candidate.row_revision = current_row.row_revision \ WHERE current_row.deleted = FALSE" ))) .bind(¤t_ids) .bind(¤t_revisions) .fetch_one(pool) .await .map_err(|error| Status::internal(format!("Current search count validation failed: {error}")))? }; let archive_count = if archives.is_empty() { 0 } else { let definition_ids = archives.iter().map(|item| item.0).collect::>(); let row_ids = archives.iter().map(|item| item.1).collect::>(); let versions = archives.iter().map(|item| item.2).collect::>(); sqlx::query_scalar::<_, i64>( r#"SELECT COUNT(*) FROM UNNEST($1::BIGINT[], $2::BIGINT[], $3::BIGINT[]) AS requested(table_definition_id, source_row_id, version_number) JOIN table_row_archives archive ON archive.table_definition_id = requested.table_definition_id AND archive.source_row_id = requested.source_row_id AND archive.version_number = requested.version_number JOIN table_definitions definition ON definition.id = archive.table_definition_id JOIN schemas owner ON owner.id = definition.schema_id WHERE definition.table_name = $4 AND definition.deleted = FALSE AND (owner.name = $5 OR definition.is_global = TRUE)"#, ) .bind(&definition_ids) .bind(&row_ids) .bind(&versions) .bind(table_name) .bind(profile_name) .fetch_one(pool) .await .map_err(|error| Status::internal(format!("Archived search count validation failed: {error}")))? }; u64::try_from(current_count.saturating_add(archive_count)) .map_err(|_| Status::internal("Search count was negative")) } struct ProfileIndex { index: Index, reader: IndexReader, fields: SchemaFields, } impl ProfileIndex { fn open(path: &Path) -> Result { let index = Index::open_in_dir(path) .map_err(|e| Status::internal(format!("Failed to open index: {}", e)))?; register_tokenizers(&index) .map_err(|e| Status::internal(format!("Failed to register tokenizers: {}", e)))?; let reader = index .reader_builder() .reload_policy(ReloadPolicy::OnCommitWithDelay) .try_into() .map_err(|e| Status::internal(format!("Failed to build index reader: {}", e)))?; let fields = SchemaFields::from(&index.schema()).map_err(|e| { Status::internal(format!( "Search index schema mismatch. Delete the stale index and create it again: {}", e )) })?; Ok(Self { index, reader, fields, }) } } #[derive(Debug)] struct NormalizedSearchRequest { profile_name: String, table_name: Option, free_query: String, must: Vec, limit: Option, offset: usize, order: Option, version_scope: SearchVersionScope, } #[derive(Debug)] struct NormalizedSearchOrder { column: String, direction: SearchOrderDirection, } #[derive(Debug)] struct NormalizedColumnConstraint { column: String, query: String, mode: ConstraintMode, } fn normalize_constraint_value(query: &str, field_type: &str) -> Result { canonical_exact_search_value(query, field_type).map_err(Status::invalid_argument) } fn public_search_system_type(column: &str) -> Option<&'static str> { match column { "id" | "row_revision" => Some("bigint"), "created_at" => Some("instant"), _ => None, } } #[derive(Clone, Debug)] struct SearchCandidate { score: f32, row_id: i64, row_revision: i64, table_name: String, row_key: String, } impl SearchCandidate { fn validation_key(&self) -> (String, Option) { if parse_archived_search_row_key(&self.row_key).is_some() { (self.row_key.clone(), None) } else { (self.row_key.clone(), Some(self.row_revision)) } } } struct CandidateSegmentFields { pg_ids: tantivy::columnar::Column, revisions: tantivy::columnar::Column, archived_flags: tantivy::columnar::Column, definition_ids: tantivy::columnar::Column, versions: tantivy::columnar::Column, table_names: tantivy::columnar::StrColumn, } fn candidate_segment_fields( searcher: &tantivy::Searcher, segment_ord: u32, ) -> Result { let segment = searcher.segment_reader(segment_ord); let fast_fields = segment.fast_fields(); let u64_column = |field_name: &str| { fast_fields.u64(field_name).map_err(|error| { Status::internal(format!( "Search candidate fast field '{field_name}' failed: {error}" )) }) }; let table_names = fast_fields .str(F_TABLE_NAME) .map_err(|error| { Status::internal(format!( "Search candidate fast field '{F_TABLE_NAME}' failed: {error}" )) })? .ok_or_else(|| { Status::internal(format!( "Search candidate fast field '{F_TABLE_NAME}' is missing" )) })?; Ok(CandidateSegmentFields { pg_ids: u64_column(F_PG_ID)?, revisions: u64_column(F_ROW_REVISION)?, archived_flags: u64_column(F_IS_ARCHIVED)?, definition_ids: u64_column(F_TABLE_DEFINITION_ID)?, versions: u64_column(F_VERSION_NUMBER)?, table_names, }) } fn search_candidates_from_documents( searcher: &tantivy::Searcher, documents: Vec<(f32, tantivy::DocAddress)>, ) -> Result, Status> { let mut fields_by_segment = HashMap::new(); for (_, address) in &documents { if !fields_by_segment.contains_key(&address.segment_ord) { fields_by_segment.insert( address.segment_ord, candidate_segment_fields(searcher, address.segment_ord)?, ); } } let mut candidates = Vec::with_capacity(documents.len()); for (score, address) in documents { let fields = fields_by_segment.get(&address.segment_ord).ok_or_else(|| { Status::internal("Search candidate segment fields were not loaded") })?; let read_u64 = |column: &tantivy::columnar::Column, field_name: &str| { column.first(address.doc_id).ok_or_else(|| { Status::internal(format!( "Search candidate document has no '{field_name}'" )) }) }; let row_id = i64::try_from(read_u64(&fields.pg_ids, F_PG_ID)?) .map_err(|_| Status::internal("Search candidate row id is out of range"))?; let row_revision = i64::try_from(read_u64(&fields.revisions, F_ROW_REVISION)?) .map_err(|_| Status::internal("Search candidate revision is out of range"))?; let archived = read_u64(&fields.archived_flags, F_IS_ARCHIVED)? != 0; let table_name_ord = fields .table_names .ords() .first(address.doc_id) .ok_or_else(|| Status::internal("Search candidate document has no table name"))?; let mut table_name = String::new(); if !fields .table_names .ord_to_str(table_name_ord, &mut table_name) .map_err(|error| { Status::internal(format!("Search candidate table name read failed: {error}")) })? { return Err(Status::internal( "Search candidate table name dictionary entry is missing", )); } let row_key = if archived { let definition_id = i64::try_from(read_u64( &fields.definition_ids, F_TABLE_DEFINITION_ID, )?) .map_err(|_| Status::internal("Search candidate table id is out of range"))?; let version = i64::try_from(read_u64(&fields.versions, F_VERSION_NUMBER)?) .map_err(|_| Status::internal("Search candidate version is out of range"))?; common::search::archived_search_row_key(definition_id, row_id, version) } else { search_row_key(&table_name, row_id) }; candidates.push(SearchCandidate { score, row_id, row_revision, table_name, row_key, }); } Ok(candidates) } impl NormalizedSearchRequest { fn has_input(&self) -> bool { !self.free_query.is_empty() || !self.must.is_empty() } } fn profile_index( cache: &Mutex>>, profile_name: &str, path: &Path, ) -> Result, Status> { { let cache_guard = cache .lock() .map_err(|_| Status::internal("Profile index cache lock poisoned"))?; if let Some(index) = cache_guard.get(profile_name) { return Ok(index.clone()); } } let opened = Arc::new(ProfileIndex::open(path)?); let mut cache_guard = cache .lock() .map_err(|_| Status::internal("Profile index cache lock poisoned"))?; if let Some(index) = cache_guard.get(profile_name) { return Ok(index.clone()); } cache_guard.insert(profile_name.to_string(), opened.clone()); Ok(opened) } fn validate_identifier(value: &str, field_name: &str) -> Result<(), Status> { let mut chars = value.chars(); let Some(first) = chars.next() else { return Err(Status::invalid_argument(format!( "{field_name} must not be empty" ))); }; if !(first.is_ascii_alphabetic() || first == '_') || !chars.all(|ch| ch.is_ascii_alphanumeric() || ch == '_') { return Err(Status::invalid_argument(format!( "{field_name} contains invalid characters" ))); } Ok(()) } fn validate_search_column(value: &str) -> Result<(), Status> { if value.is_empty() { return Err(Status::invalid_argument( "constraint.column must not be empty", )); } if value.chars().any(|ch| ch.is_control() || ch == '\0') { return Err(Status::invalid_argument( "constraint.column contains invalid characters", )); } Ok(()) } fn qualify_profile_table(profile_name: &str, table_name: &str) -> String { format!("\"{}\".\"{}\"", profile_name, table_name) } async fn profile_exists(pool: &PgPool, profile_name: &str) -> Result { let exists = sqlx::query_scalar::<_, bool>("SELECT EXISTS(SELECT 1 FROM schemas WHERE name = $1)") .bind(profile_name) .fetch_one(pool) .await .map_err(|e| Status::internal(format!("Profile lookup failed: {}", e)))?; Ok(exists) } async fn table_exists(pool: &PgPool, profile_name: &str, table_name: &str) -> Result { 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) ) "#, ) .bind(profile_name) .bind(table_name) .fetch_one(pool) .await .map_err(|e| Status::internal(format!("Table lookup failed: {}", e)))?; Ok(exists) } async fn qualified_visible_table( pool: &PgPool, profile_name: &str, table_name: &str, ) -> Result { let storage_schema = sqlx::query_scalar::<_, String>( r#"SELECT owner.name FROM table_definitions definition JOIN schemas owner ON owner.id = definition.schema_id WHERE definition.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) .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)) } fn normalize_request(req: SearchRequest) -> Result { let profile_name = req.profile_name.trim(); if profile_name.is_empty() { return Err(Status::invalid_argument("profile_name is required")); } validate_identifier(profile_name, "profile_name")?; 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")?; Some(table_name.to_string()) } _ => None, }; let free_query = req.free_query.trim().to_string(); let mut must = Vec::new(); for constraint in req.must { let column = constraint.column.trim(); validate_search_column(column)?; let query = constraint.query.trim(); if query.is_empty() { return Err(Status::invalid_argument( "constraint.query must not be empty", )); } must.push(NormalizedColumnConstraint { column: column.to_string(), query: query.to_string(), mode: constraint_mode_from_proto(constraint.mode), }); } let limit = req .limit .map(|value| (value as usize).min(HARD_RESULT_LIMIT)); let offset = req.offset.unwrap_or_default() as usize; let order = req .order .map(|order| { let column = order.column.trim(); if column.is_empty() { return Err(Status::invalid_argument("order.column must not be empty")); } validate_search_column(column)?; let direction = SearchOrderDirection::try_from(order.direction).map_err(|_| { Status::invalid_argument("order.direction must be ASC or DESC") })?; if direction == SearchOrderDirection::Unspecified { return Err(Status::invalid_argument( "order.direction must be ASC or DESC", )); } Ok(NormalizedSearchOrder { column: column.to_string(), direction, }) }) .transpose()?; let version_scope = SearchVersionScope::try_from(req.version_scope) .map_err(|_| Status::invalid_argument("version_scope is invalid"))?; if order.is_some() && table_name.is_none() { return Err(Status::invalid_argument( "table_name is required when order is specified", )); } if order.is_some() && version_scope != SearchVersionScope::Current { return Err(Status::invalid_argument( "ordering archived search results is not supported", )); } Ok(NormalizedSearchRequest { profile_name: profile_name.to_string(), table_name, free_query, must, limit, offset, order, version_scope, }) } fn constraint_mode_from_proto(raw_mode: i32) -> ConstraintMode { match raw_mode { 2 => ConstraintMode::Exact, _ => ConstraintMode::Fuzzy, } } async fn resolve_constraints( pool: &PgPool, profile_name: &str, table_filter: Option<&str>, must: &[NormalizedColumnConstraint], ) -> Result, Status> { let mut resolved = Vec::with_capacity(must.len()); for constraint in must { let targets = resolve_constraint_targets( pool, profile_name, table_filter, &constraint.column, &constraint.query, constraint.mode, ) .await?; resolved.push(SearchConstraint { targets, mode: constraint.mode, }); } Ok(resolved) } async fn resolve_constraint_targets( pool: &PgPool, profile_name: &str, table_filter: Option<&str>, column: &str, query: &str, mode: ConstraintMode, ) -> Result, Status> { let rows = if let Some(table_name) = table_filter { sqlx::query( r#" SELECT td.table_name, tdc.physical_name, tdc.field_type FROM schemas s JOIN table_definitions td ON td.schema_id = s.id JOIN table_definition_columns tdc ON tdc.table_definition_id = td.id WHERE td.id = ( SELECT visible.id FROM table_definitions visible JOIN schemas owner ON owner.id = visible.schema_id WHERE visible.table_name = $2 AND visible.deleted = FALSE AND (owner.name = $1 OR visible.is_global = TRUE) ORDER BY visible.is_global ASC LIMIT 1 ) AND tdc.display_name = $3 "#, ) .bind(profile_name) .bind(table_name) .bind(column) .fetch_all(pool) .await } else { sqlx::query( r#" SELECT td.table_name, tdc.physical_name, tdc.field_type FROM schemas s JOIN table_definitions td ON td.schema_id = s.id JOIN table_definition_columns tdc ON tdc.table_definition_id = td.id WHERE (s.name = $1 OR td.is_global = TRUE) AND td.deleted = FALSE AND tdc.display_name = $2 "#, ) .bind(profile_name) .bind(column) .fetch_all(pool) .await } .map_err(|e| Status::internal(format!("Column mapping lookup failed: {}", e)))?; let rows = if rows.is_empty() { let Some(field_type) = public_search_system_type(column) else { return Err(Status::invalid_argument(format!( "Column alias '{}' was not found{}", column, table_filter .map(|table_name| format!(" in table '{}'", table_name)) .unwrap_or_default() ))); }; let table_names = if let Some(table_name) = table_filter { vec![table_name.to_string()] } else { sqlx::query_scalar::<_, String>( r#"SELECT DISTINCT definition.table_name FROM table_definitions definition JOIN schemas owner ON owner.id = definition.schema_id WHERE definition.deleted = FALSE AND (owner.name = $1 OR definition.is_global = TRUE)"#, ) .bind(profile_name) .fetch_all(pool) .await .map_err(|error| Status::internal(format!( "System-column search table lookup failed: {error}" )))? }; let query = if mode == ConstraintMode::Exact { normalize_constraint_value(query, field_type)? } else { query.to_string() }; return Ok(table_names .into_iter() .map(|table_name| SearchConstraintTarget { table_name: table_filter.is_none().then_some(table_name), column: column.to_string(), query: query.clone(), }) .collect()); } else { rows }; let mut seen = HashSet::new(); let mut targets = Vec::new(); for row in rows { let table_name: String = row .try_get("table_name") .map_err(|e| Status::internal(format!("Column mapping table read failed: {}", e)))?; let physical_name: String = row .try_get("physical_name") .map_err(|e| Status::internal(format!("Column mapping physical read failed: {}", e)))?; let field_type: String = row .try_get("field_type") .map_err(|e| Status::internal(format!("Column mapping type read failed: {e}")))?; let key = (table_name.clone(), physical_name.clone()); if seen.insert(key) { targets.push(SearchConstraintTarget { table_name: if table_filter.is_some() { None } else { Some(table_name) }, column: physical_name, query: if mode == ConstraintMode::Exact { normalize_constraint_value(query, &field_type)? } else { query.to_string() }, }); } } Ok(targets) } async fn table_physical_to_display_map( pool: &PgPool, profile_name: &str, table_name: &str, ) -> Result, Status> { let rows = sqlx::query( r#" SELECT tdc.physical_name, tdc.display_name FROM schemas s JOIN table_definitions td ON td.schema_id = s.id JOIN table_definition_columns tdc ON tdc.table_definition_id = td.id WHERE td.id = ( SELECT visible.id FROM table_definitions visible JOIN schemas owner ON owner.id = visible.schema_id WHERE visible.table_name = $2 AND visible.deleted = FALSE AND (owner.name = $1 OR visible.is_global = TRUE) ORDER BY visible.is_global ASC LIMIT 1 ) "#, ) .bind(profile_name) .bind(table_name) .fetch_all(pool) .await .map_err(|e| Status::internal(format!("Column mapping lookup failed: {}", e)))?; let mut mapping = HashMap::with_capacity(rows.len()); for row in rows { let physical_name: String = row .try_get("physical_name") .map_err(|e| Status::internal(format!("Column mapping physical read failed: {}", e)))?; let display_name: String = row .try_get("display_name") .map_err(|e| Status::internal(format!("Column mapping display read failed: {}", e)))?; mapping.insert(physical_name, display_name); } Ok(mapping) } async fn table_internal_column_names( pool: &PgPool, profile_name: &str, table_name: &str, ) -> Result, Status> { let rows = sqlx::query_scalar::<_, String>( r#" SELECT link.version_column_name FROM schemas s JOIN table_definitions td ON td.schema_id = s.id LEFT JOIN table_definition_links link ON link.source_table_id = td.id WHERE td.id = ( SELECT visible.id FROM table_definitions visible JOIN schemas owner ON owner.id = visible.schema_id WHERE visible.table_name = $2 AND visible.deleted = FALSE AND (owner.name = $1 OR visible.is_global = TRUE) ORDER BY visible.is_global ASC LIMIT 1 ) AND link.version_column_name IS NOT NULL "#, ) .bind(profile_name) .bind(table_name) .fetch_all(pool) .await .map_err(|e| Status::internal(format!("Internal column lookup failed: {e}")))?; let mut internal = internal_column_names() .map(str::to_string) .collect::>(); internal.extend(rows); Ok(internal) } async fn table_row_display_columns( pool: &PgPool, profile_name: &str, table_name: &str, ) -> Result, Status> { sqlx::query_scalar( r#" SELECT td.row_display_columns FROM schemas s JOIN table_definitions td ON td.schema_id = s.id WHERE td.id = ( SELECT visible.id FROM table_definitions visible JOIN schemas owner ON owner.id = visible.schema_id WHERE visible.table_name = $2 AND visible.deleted = FALSE AND (owner.name = $1 OR visible.is_global = TRUE) ORDER BY visible.is_global ASC LIMIT 1 ) "#, ) .bind(profile_name) .bind(table_name) .fetch_one(pool) .await .map_err(|e| Status::internal(format!("Row display columns lookup failed: {}", e))) } fn remap_json_to_display_names( value: serde_json::Value, physical_to_display: &HashMap, internal_columns: &HashSet, ) -> Result { match value { serde_json::Value::Object(object) => { let mut remapped = serde_json::Map::with_capacity(object.len()); for (key, value) in object { if internal_columns.contains(&key) { continue; } let final_key = match physical_to_display.get(&key) { Some(display_name) => display_name.clone(), None if is_system_column(&key) => key, None => { return Err(Status::failed_precondition( "A table column has no public alias mapping", )); } }; remapped.insert(final_key, value); } Ok(serde_json::Value::Object(remapped)) } other => Ok(other), } } /// One value per display column, positionally aligned with them, so a column /// that is NULL for this row stays visible as an empty slot. fn row_display_values(value: &serde_json::Value, columns: &[String]) -> Vec { columns .iter() .map(|column| match value.get(column) { Some(serde_json::Value::String(value)) => value.clone(), Some(serde_json::Value::Number(value)) => value.to_string(), Some(serde_json::Value::Bool(value)) => value.to_string(), _ => String::new(), }) .collect() } enum ResolvedOrderColumn { Position, Column(String), } fn order_direction_sql(direction: SearchOrderDirection) -> &'static str { match direction { SearchOrderDirection::Asc => "ASC", SearchOrderDirection::Desc => "DESC", SearchOrderDirection::Unspecified => unreachable!("order direction is normalized"), } } async fn resolve_order_column( pool: &PgPool, profile_name: &str, table_name: &str, requested_column: &str, ) -> Result { if requested_column.eq_ignore_ascii_case("position") { return Ok(ResolvedOrderColumn::Position); } // Sorting by "the display column" means the first one: it is the part // callers read left to right. A table with none sorts by id instead. let requested_column = if requested_column.eq_ignore_ascii_case("row_display_columns") { table_row_display_columns(pool, profile_name, table_name) .await? .into_iter() .next() .unwrap_or_else(|| "id".to_string()) } else { requested_column.to_string() }; let physical_to_display = table_physical_to_display_map(pool, profile_name, table_name).await?; let physical_column = physical_to_display .iter() .find(|(_, display)| display.eq_ignore_ascii_case(&requested_column)) .map(|(physical, _)| physical.clone()) .or_else(|| { (is_system_column(&requested_column) && !is_internal_column(&requested_column) && !physical_to_display.contains_key(&requested_column)) .then(|| requested_column.clone()) }) .ok_or_else(|| { Status::invalid_argument(format!( "Column alias '{}' was not found in table '{}.{}'", requested_column, profile_name, table_name )) })?; let physical_column = sqlx::query_scalar::<_, String>( r#" SELECT column_name FROM information_schema.columns WHERE table_schema = ( SELECT owner.name FROM table_definitions definition JOIN schemas owner ON owner.id = definition.schema_id WHERE definition.table_name = $2 AND definition.deleted = FALSE AND (owner.name = $1 OR definition.is_global = TRUE) ORDER BY definition.is_global ASC LIMIT 1 ) AND table_name = $2 AND LOWER(column_name) = LOWER($3) "#, ) .bind(profile_name) .bind(table_name) .bind(&physical_column) .fetch_optional(pool) .await .map_err(|e| Status::internal(format!("Order column lookup failed: {}", e)))?; let Some(physical_column) = physical_column else { return Err(Status::invalid_argument(format!( "Order column '{}' was not found in table '{}.{}'", requested_column, profile_name, table_name ))); }; Ok(ResolvedOrderColumn::Column(physical_column)) } fn order_clause(column: &ResolvedOrderColumn, direction: SearchOrderDirection) -> String { let direction = order_direction_sql(direction); match column { ResolvedOrderColumn::Position => { format!("picker_position {}", direction) } ResolvedOrderColumn::Column(column) => { let quoted_column = format!("\"{}\"", column.replace('"', "\"\"")); format!("{} {} NULLS LAST, id ASC", quoted_column, direction) } } } fn ranked_order_clause( column: &ResolvedOrderColumn, direction: SearchOrderDirection, ) -> String { let direction = order_direction_sql(direction); match column { ResolvedOrderColumn::Position => { format!("candidate.candidate_group DESC, positioned.picker_position {}", direction) } ResolvedOrderColumn::Column(column) => { let quoted_column = format!("\"{}\"", column.replace('"', "\"\"")); format!( "candidate.candidate_group DESC, positioned.{} {} NULLS LAST, positioned.id ASC", quoted_column, direction ) } } } fn relevance_group(score: f32, best_score: f32) -> i32 { if best_score <= 0.0 || !best_score.is_finite() || !score.is_finite() { return 0; } ((score.max(0.0) / best_score / SEARCH_SCORE_GROUP_WIDTH).floor() as i32).clamp(0, 3) } async fn fetch_ordered_rows( pool: &PgPool, profile_name: &str, table_name: &str, order: Option<&NormalizedSearchOrder>, limit: usize, offset: usize, ) -> Result, Status> { let physical_to_display = table_physical_to_display_map(pool, profile_name, table_name).await?; let internal_columns = table_internal_column_names(pool, profile_name, table_name).await?; let account_projection = AccountPathProjection::load( pool, profile_name, table_name, &physical_to_display, ) .await?; let display_columns = account_projection.row_display_columns( &table_row_display_columns(pool, profile_name, table_name).await?, ); let (resolved_order, direction) = match order { Some(order) => ( resolve_order_column(pool, profile_name, table_name, &order.column).await?, order.direction, ), None => (ResolvedOrderColumn::Position, SearchOrderDirection::Desc), }; let qualified_table = qualified_visible_table(pool, profile_name, table_name).await?; let sql = format!( "WITH positioned AS (\ SELECT t.*, ROW_NUMBER() OVER (ORDER BY id ASC) AS picker_position \ FROM {} t WHERE deleted = FALSE\ ) \ SELECT id, to_jsonb(positioned) - 'picker_position' AS data, picker_position \ FROM positioned ORDER BY {} LIMIT $1 OFFSET $2", qualified_table, order_clause(&resolved_order, direction), ); let rows = sqlx::query(AssertSqlSafe(sql)) .bind(limit as i64) .bind(offset as i64) .fetch_all(pool) .await .map_err(|e| Status::internal(format!("DB query for default results failed: {}", e)))?; rows .into_iter() .map(|row| -> Result { let id: i64 = row .try_get("id") .map_err(|error| Status::internal(format!("Search id read failed: {}", error)))?; let json_data: serde_json::Value = row.try_get("data").map_err(|error| { Status::internal(format!("Search row read failed: {}", error)) })?; let position: i64 = row.try_get("picker_position").map_err(|error| { Status::internal(format!("Search position read failed: {}", error)) })?; let version_number = json_data .get("version") .and_then(|value| value.as_i64()) .unwrap_or(0); let mut json_data = remap_json_to_display_names( json_data, &physical_to_display, &internal_columns, )?; account_projection.apply(id, &mut json_data)?; let row_display_values = row_display_values(&json_data, &display_columns); Ok(Hit { id, score: 0.0, content_json: json_data.to_string(), table_name: table_name.to_string(), row_display_values, row_display_columns: display_columns.clone(), position: u64::try_from(position).ok(), version_number, archived: false, }) }) .collect() } async fn validate_search_candidates( pool: &PgPool, profile_name: &str, candidates: &[SearchCandidate], ) -> Result)>, Status> { let mut valid_keys = HashSet::new(); let mut current_by_table: HashMap<&str, Vec<(i64, i64)>> = HashMap::new(); let mut archives_by_table: HashMap<&str, Vec<(i64, i64, i64)>> = HashMap::new(); for candidate in candidates { if let Some(archive) = parse_archived_search_row_key(&candidate.row_key) { archives_by_table .entry(&candidate.table_name) .or_default() .push(archive); } else { current_by_table .entry(&candidate.table_name) .or_default() .push((candidate.row_id, candidate.row_revision)); } } for (table_name, rows) in current_by_table { let qualified_table = qualified_visible_table(pool, profile_name, table_name).await?; let ids = rows.iter().map(|item| item.0).collect::>(); let revisions = rows.iter().map(|item| item.1).collect::>(); let sql = format!( "SELECT candidate.id, candidate.row_revision \ FROM UNNEST($1::BIGINT[], $2::BIGINT[]) \ AS candidate(id, row_revision) \ JOIN {qualified_table} current_row \ ON current_row.id = candidate.id \ AND current_row.row_revision = candidate.row_revision \ WHERE current_row.deleted = FALSE" ); let valid_rows = sqlx::query(AssertSqlSafe(sql)) .bind(&ids) .bind(&revisions) .fetch_all(pool) .await .map_err(|error| { Status::internal(format!("Current search candidate validation failed: {error}")) })?; for row in valid_rows { let row_id: i64 = row.try_get("id").map_err(|error| { Status::internal(format!("Current search id read failed: {error}")) })?; let revision: i64 = row.try_get("row_revision").map_err(|error| { Status::internal(format!("Current search revision read failed: {error}")) })?; valid_keys.insert((search_row_key(table_name, row_id), Some(revision))); } } for (table_name, archives) in archives_by_table { let definition_ids = archives.iter().map(|item| item.0).collect::>(); let row_ids = archives.iter().map(|item| item.1).collect::>(); let versions = archives.iter().map(|item| item.2).collect::>(); let rows = sqlx::query( r#"SELECT archive.table_definition_id, archive.source_row_id, archive.version_number FROM UNNEST($1::BIGINT[], $2::BIGINT[], $3::BIGINT[]) AS requested(table_definition_id, source_row_id, version_number) JOIN table_row_archives archive ON archive.table_definition_id = requested.table_definition_id AND archive.source_row_id = requested.source_row_id AND archive.version_number = requested.version_number JOIN table_definitions definition ON definition.id = archive.table_definition_id JOIN schemas owner ON owner.id = definition.schema_id WHERE definition.table_name = $4 AND definition.deleted = FALSE AND (owner.name = $5 OR definition.is_global = TRUE)"#, ) .bind(&definition_ids) .bind(&row_ids) .bind(&versions) .bind(table_name) .bind(profile_name) .fetch_all(pool) .await .map_err(|error| { Status::internal(format!("Archived search candidate validation failed: {error}")) })?; for row in rows { let definition_id: i64 = row.try_get("table_definition_id").map_err(|error| { Status::internal(format!("Archived table id read failed: {error}")) })?; let row_id: i64 = row.try_get("source_row_id").map_err(|error| { Status::internal(format!("Archived row id read failed: {error}")) })?; let version: i64 = row.try_get("version_number").map_err(|error| { Status::internal(format!("Archived version read failed: {error}")) })?; valid_keys.insert(( common::search::archived_search_row_key(definition_id, row_id, version), None, )); } } Ok(valid_keys) } async fn run_search( pool: &PgPool, profile: &ProfileIndex, profile_name: &str, table_filter: Option<&str>, free_query: &str, must: &[SearchConstraint], limit: usize, offset: usize, order: Option<&NormalizedSearchOrder>, version_scope: SearchVersionScope, ) -> Result, Status> { let master_query = build_master_query( &profile.index, &profile.fields, free_query, must, table_filter, version_scope, )?; let searcher = profile.reader.searcher(); let num_docs = searcher.num_docs() as usize; if num_docs == 0 { return Ok(Vec::new()); } let requested_candidates = offset.saturating_add(limit).max(1); let mut window_limit = if order.is_some() { num_docs } else { requested_candidates.min(num_docs) }; let candidates = loop { let top_docs = searcher .search( &*master_query, &TopDocs::with_limit(window_limit).order_by_score(), ) .map_err(|e| Status::internal(format!("Search failed: {}", e)))?; if top_docs.is_empty() { return Ok(Vec::new()); } let collected_docs = top_docs.len(); let best_score = top_docs.first().map(|(score, _)| *score).unwrap_or_default(); let score_floor = if best_score > 0.0 { best_score * SEARCH_SCORE_RELATIVE_FLOOR } else { 0.0 }; let reached_score_floor = top_docs .last() .is_some_and(|(score, _)| *score < score_floor); let eligible_docs = top_docs .into_iter() .filter(|(score, _)| *score >= score_floor) .collect::>(); let candidates = search_candidates_from_documents(&searcher, eligible_docs)?; if order.is_some() { break candidates; } let valid_keys = validate_search_candidates(pool, profile_name, &candidates).await?; let valid_count = candidates .iter() .filter(|candidate| valid_keys.contains(&candidate.validation_key())) .count(); if valid_count >= requested_candidates || window_limit == num_docs || reached_score_floor || collected_docs < window_limit { break candidates .into_iter() .filter(|candidate| valid_keys.contains(&candidate.validation_key())) .skip(offset) .take(limit) .collect(); } window_limit = window_limit.saturating_mul(2).min(num_docs); }; if candidates.is_empty() { return Ok(Vec::new()); } if let Some(order) = order { let table_name = table_filter.expect("ordered searches require a normalized table filter"); return fetch_ordered_candidate_rows( pool, profile_name, table_name, &candidates, order, limit, offset, ) .await; } let mut rows_by_table: HashMap> = HashMap::new(); for candidate in &candidates { if parse_archived_search_row_key(&candidate.row_key).is_some() { continue; } rows_by_table .entry(candidate.table_name.clone()) .or_default() .push((candidate.row_id, candidate.row_revision)); } let mut content_map: HashMap, Vec, i64, bool)> = HashMap::new(); for (table_name, candidate_rows) in rows_by_table { validate_identifier(&table_name, "table_name")?; let physical_to_display = table_physical_to_display_map(pool, profile_name, &table_name).await?; let internal_columns = table_internal_column_names(pool, profile_name, &table_name).await?; let account_projection = AccountPathProjection::load( pool, profile_name, &table_name, &physical_to_display, ) .await?; let display_columns = account_projection.row_display_columns( &table_row_display_columns(pool, profile_name, &table_name).await?, ); let qualified_table = qualified_visible_table(pool, profile_name, &table_name).await?; let pg_ids = candidate_rows.iter().map(|item| item.0).collect::>(); let revisions_by_id = candidate_rows.into_iter().collect::>(); let sql = format!( "SELECT id, to_jsonb(t) AS data FROM {} t WHERE deleted = FALSE AND id = ANY($1)", qualified_table ); let rows = sqlx::query(AssertSqlSafe(sql)) .bind(&pg_ids) .fetch_all(pool) .await .map_err(|e| Status::internal(format!("Database query failed: {}", e)))?; for row in rows { let id: i64 = row.try_get("id").unwrap_or_default(); let json_data: serde_json::Value = row.try_get("data").unwrap_or_default(); let row_revision = json_data .get("row_revision") .and_then(|value| value.as_i64()); if revisions_by_id.get(&id).copied() != row_revision { continue; } let version = json_data.get("version").and_then(|value| value.as_i64()).unwrap_or(0); let mut json_data = remap_json_to_display_names( json_data, &physical_to_display, &internal_columns, )?; account_projection.apply(id, &mut json_data)?; let display_values = row_display_values(&json_data, &display_columns); content_map.insert(search_row_key(&table_name, id), ( json_data.to_string(), display_values, display_columns.clone(), version, false, )); } } let mut archives_by_table: HashMap> = HashMap::new(); for candidate in &candidates { if let Some((table_definition_id, row_id, version)) = parse_archived_search_row_key(&candidate.row_key) { archives_by_table .entry(candidate.table_name.clone()) .or_default() .push((table_definition_id, row_id, version)); } } for (table_name, archives) in archives_by_table { let physical_to_display = table_physical_to_display_map(pool, profile_name, &table_name).await?; let internal_columns = table_internal_column_names(pool, profile_name, &table_name).await?; let account_projection = AccountPathProjection::load( pool, profile_name, &table_name, &physical_to_display, ) .await?; let display_columns = account_projection.row_display_columns( &table_row_display_columns(pool, profile_name, &table_name).await?, ); let table_definition_ids = archives.iter().map(|item| item.0).collect::>(); let row_ids = archives.iter().map(|item| item.1).collect::>(); let versions = archives.iter().map(|item| item.2).collect::>(); let rows = sqlx::query( r#"SELECT archive.table_definition_id, archive.source_row_id, archive.version_number, archive.snapshot FROM UNNEST($1::BIGINT[], $2::BIGINT[], $3::BIGINT[]) AS requested(table_definition_id, source_row_id, version_number) JOIN table_row_archives archive ON archive.table_definition_id = requested.table_definition_id AND archive.source_row_id = requested.source_row_id AND archive.version_number = requested.version_number JOIN table_definitions definition ON definition.id = archive.table_definition_id JOIN schemas owner ON owner.id = definition.schema_id WHERE definition.table_name = $4 AND definition.deleted = FALSE AND (owner.name = $5 OR definition.is_global = TRUE)"#, ) .bind(&table_definition_ids) .bind(&row_ids) .bind(&versions) .bind(&table_name) .bind(profile_name) .fetch_all(pool) .await .map_err(|error| Status::internal(format!( "Archived search row lookup failed: {error}" )))?; for row in rows { let table_definition_id: i64 = row.try_get("table_definition_id").map_err(|error| { Status::internal(format!("Archived table id read failed: {error}")) })?; let row_id: i64 = row.try_get("source_row_id").map_err(|error| { Status::internal(format!("Archived row id read failed: {error}")) })?; let version: i64 = row.try_get("version_number").map_err(|error| { Status::internal(format!("Archived version read failed: {error}")) })?; let snapshot: serde_json::Value = row.try_get("snapshot").map_err(|error| { Status::internal(format!("Archived snapshot read failed: {error}")) })?; let row_key = common::search::archived_search_row_key( table_definition_id, row_id, version, ); let mut json_data = remap_json_to_display_names( snapshot, &physical_to_display, &internal_columns, )?; account_projection.apply(row_id, &mut json_data)?; let display_values = row_display_values(&json_data, &display_columns); content_map.insert(row_key, ( json_data.to_string(), display_values, display_columns.clone(), version, true, )); } } let hits = candidates .into_iter() .filter_map(|candidate| { content_map .get(&candidate.row_key) .map(|(content_json, row_display_values, row_display_columns, version_number, archived)| Hit { id: candidate.row_id, score: candidate.score, content_json: content_json.clone(), table_name: candidate.table_name, row_display_values: row_display_values.clone(), row_display_columns: row_display_columns.clone(), position: None, version_number: *version_number, archived: *archived, }) }) .collect::>(); Ok(hits) } async fn fetch_ordered_candidate_rows( pool: &PgPool, profile_name: &str, table_name: &str, candidates: &[SearchCandidate], order: &NormalizedSearchOrder, limit: usize, offset: usize, ) -> Result, Status> { let physical_to_display = table_physical_to_display_map(pool, profile_name, table_name).await?; let internal_columns = table_internal_column_names(pool, profile_name, table_name).await?; let account_projection = AccountPathProjection::load( pool, profile_name, table_name, &physical_to_display, ) .await?; let display_columns = account_projection.row_display_columns( &table_row_display_columns(pool, profile_name, table_name).await?, ); let resolved_order = resolve_order_column(pool, profile_name, table_name, &order.column).await?; let qualified_table = qualified_visible_table(pool, profile_name, table_name).await?; let sql = format!( "WITH positioned AS (\ SELECT t.*, ROW_NUMBER() OVER (ORDER BY id ASC) AS picker_position \ FROM {} t WHERE deleted = FALSE\ ) \ SELECT positioned.id, to_jsonb(positioned) - 'picker_position' AS data, \ picker_position, candidate_score \ FROM positioned \ JOIN UNNEST($1::BIGINT[], $2::BIGINT[], $3::REAL[], $4::INTEGER[]) \ AS candidate(candidate_id, candidate_revision, candidate_score, candidate_group) \ ON candidate_id = positioned.id \ AND candidate_revision = positioned.row_revision \ ORDER BY {} LIMIT $5 OFFSET $6", qualified_table, ranked_order_clause(&resolved_order, order.direction), ); let ids = candidates .iter() .map(|candidate| candidate.row_id) .collect::>(); let scores = candidates .iter() .map(|candidate| candidate.score) .collect::>(); let revisions = candidates .iter() .map(|candidate| candidate.row_revision) .collect::>(); let best_score = scores .iter() .copied() .max_by(f32::total_cmp) .unwrap_or_default(); let groups = scores .iter() .map(|score| relevance_group(*score, best_score)) .collect::>(); let rows = sqlx::query(AssertSqlSafe(sql)) .bind(&ids) .bind(&revisions) .bind(&scores) .bind(&groups) .bind(limit as i64) .bind(offset as i64) .fetch_all(pool) .await .map_err(|e| Status::internal(format!("Ordered search query failed: {}", e)))?; rows .into_iter() .map(|row| -> Result { let id: i64 = row .try_get("id") .map_err(|error| Status::internal(format!("Search id read failed: {}", error)))?; let json_data: serde_json::Value = row.try_get("data").map_err(|error| { Status::internal(format!("Search row read failed: {}", error)) })?; let position: i64 = row.try_get("picker_position").map_err(|error| { Status::internal(format!("Search position read failed: {}", error)) })?; let score: f32 = row.try_get("candidate_score").map_err(|error| { Status::internal(format!("Search score read failed: {}", error)) })?; let version_number = json_data .get("version") .and_then(|value| value.as_i64()) .unwrap_or(0); let mut json_data = remap_json_to_display_names( json_data, &physical_to_display, &internal_columns, )?; account_projection.apply(id, &mut json_data)?; let display_values = row_display_values(&json_data, &display_columns); Ok(Hit { id, score, content_json: json_data.to_string(), table_name: table_name.to_string(), row_display_values: display_values, row_display_columns: display_columns.clone(), position: u64::try_from(position).ok(), version_number, archived: false, }) }) .collect() } #[tonic::async_trait] impl Searcher for SearcherService { async fn search( &self, request: Request, ) -> Result, Status> { self.run_rpc(request).await } async fn count( &self, request: Request, ) -> Result, Status> { self.run_count_rpc(request).await } } #[cfg(test)] mod tests { use super::*; use common::proto::komp_ac::search::{SearchOrder, SearchOrderDirection}; use common::search::create_search_schema; use tantivy::query::AllQuery; #[test] fn search_response_mapping_exposes_aliases_only() { let mappings = HashMap::from([("1".to_string(), "customer".to_string())]); let mapped = remap_json_to_display_names( serde_json::json!({"1": "Acme", "id": 4}), &mappings, &HashSet::new(), ) .unwrap(); assert_eq!(mapped, serde_json::json!({"customer": "Acme", "id": 4})); } #[test] fn account_projection_replaces_internal_ids_with_public_paths() { let projection = AccountPathProjection { paths: HashMap::from([(3, "600/10".to_string())]), account_column: Some("account".to_string()), accounts_table: false, }; let mut value = serde_json::json!({"account": 3, "name": "posting"}); projection.apply(9, &mut value).unwrap(); assert_eq!(value, serde_json::json!({"account": "600/10", "name": "posting"})); } #[test] fn accounts_search_exposes_and_displays_the_complete_path() { let projection = AccountPathProjection { paths: HashMap::from([(3, "600/10".to_string())]), account_column: None, accounts_table: true, }; let mut value = serde_json::json!({"segment": "10", "parent_account_id": 2}); projection.apply(3, &mut value).unwrap(); assert_eq!(value["account"], "600/10"); assert_eq!(projection.row_display_columns(&["segment".to_string()]), ["account"]); } #[test] fn count_collector_returns_zero_documents_for_an_empty_index() { let schema = create_search_schema(); let index = Index::create_in_ram(schema.clone()); let profile = ProfileIndex { reader: index.reader().expect("empty reader should open"), fields: SchemaFields::from(&schema).expect("search schema should match"), index, }; assert_eq!( top_documents_for_count(&profile.reader.searcher(), &AllQuery) .expect("empty count should not construct a zero-limit collector") .len(), 0, ); } #[test] fn search_response_mapping_fails_instead_of_leaking_a_real_name() { let error = remap_json_to_display_names( serde_json::json!({"1": "Acme", "2": "hidden"}), &HashMap::from([("1".to_string(), "customer".to_string())]), &HashSet::new(), ) .unwrap_err(); assert_eq!(error.code(), tonic::Code::FailedPrecondition); assert!(!error.message().contains("'2'")); } #[test] fn search_response_mapping_omits_registered_internal_columns() { let mapped = remap_json_to_display_names( serde_json::json!({ "1": 7, "1_version": 3, "account_id": 11, "account_id_version": 4, "version": 5, "id": 9 }), &HashMap::from([ ("1".to_string(), "adresar".to_string()), ("account_id".to_string(), "account".to_string()), ]), &HashSet::from([ "1_version".to_string(), "account_id_version".to_string(), "version".to_string(), ]), ) .unwrap(); assert_eq!( mapped, serde_json::json!({"adresar": 7, "account": 11, "id": 9}) ); } fn request(order: Option, table_name: Option<&str>) -> SearchRequest { SearchRequest { profile_name: "finance".to_string(), table_name: table_name.map(str::to_string), free_query: String::new(), must: Vec::new(), limit: Some(20), offset: Some(40), order, version_scope: 0, } } #[test] fn normalizes_typed_column_order() { let normalized = normalize_request(request( Some(SearchOrder { column: "created_at".to_string(), direction: SearchOrderDirection::Desc as i32, }), Some("invoice"), )) .unwrap(); let order = normalized.order.unwrap(); assert_eq!(order.column, "created_at"); assert_eq!(order.direction, SearchOrderDirection::Desc); } #[test] fn ordered_request_requires_a_table() { let result = normalize_request(request( Some(SearchOrder { column: "position".to_string(), direction: SearchOrderDirection::Desc as i32, }), None, )); assert!(result.is_err()); } #[test] fn column_order_has_stable_id_tie_breaker() { assert_eq!( order_clause( &ResolvedOrderColumn::Column("1".to_string()), SearchOrderDirection::Asc, ), "\"1\" ASC NULLS LAST, id ASC" ); } #[test] fn ranked_order_preserves_relevance_before_column_order() { assert_eq!( ranked_order_clause( &ResolvedOrderColumn::Column("created_at".to_string()), SearchOrderDirection::Desc, ), "candidate.candidate_group DESC, positioned.\"created_at\" DESC NULLS LAST, positioned.id ASC" ); } #[test] fn relevance_groups_nearby_high_scores_and_separates_low_scores() { let best_score = 8.7; assert_eq!(relevance_group(8.7, best_score), 3); assert_eq!(relevance_group(8.6, best_score), 3); assert_eq!(relevance_group(8.4, best_score), 3); assert_eq!(relevance_group(7.0, best_score), 3); assert_eq!(relevance_group(3.0, best_score), 1); } }