sql search2 added

This commit is contained in:
filipriec
2025-07-25 22:38:34 +02:00
parent 8127c7bb1b
commit 00c0a399cd
8 changed files with 654 additions and 4 deletions

View File

@@ -15,6 +15,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
"proto/tables_data.proto",
"proto/table_script.proto",
"proto/search.proto",
"proto/search2.proto",
],
&["proto"],
)?;

View File

@@ -0,0 +1,46 @@
// In common/proto/search2.proto
syntax = "proto3";
package komp_ac.search2;
service Search2 {
rpc SearchTable(Search2Request) returns (Search2Response);
}
message Search2Request {
string profile_name = 1;
string table_name = 2;
repeated ColumnFilter column_filters = 3;
optional string text_query = 4; // Optional fallback text search
optional int32 limit = 5;
optional string order_by = 6;
optional bool order_desc = 7;
}
message ColumnFilter {
string column_name = 1;
FilterType filter_type = 2;
string value = 3;
optional string value2 = 4; // For range queries
}
enum FilterType {
EQUALS = 0;
CONTAINS = 1;
STARTS_WITH = 2;
ENDS_WITH = 3;
RANGE = 4;
GREATER_THAN = 5;
LESS_THAN = 6;
IS_NULL = 7;
IS_NOT_NULL = 8;
}
message Search2Response {
message Hit {
int64 id = 1;
string content_json = 2; // No score - this is SQL-based
optional string match_info = 3; // Info about which columns matched
}
repeated Hit hits = 1;
int32 total_count = 2; // Total matching records (for pagination)
}

View File

@@ -31,6 +31,9 @@ pub mod proto {
pub mod search {
include!("proto/komp_ac.search.rs");
}
pub mod search2 {
include!("proto/komp_ac.search2.rs");
}
pub const FILE_DESCRIPTOR_SET: &[u8] =
include_bytes!("proto/descriptor.bin");
}

Binary file not shown.

View File

@@ -0,0 +1,394 @@
// This file is @generated by prost-build.
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct Search2Request {
#[prost(string, tag = "1")]
pub profile_name: ::prost::alloc::string::String,
#[prost(string, tag = "2")]
pub table_name: ::prost::alloc::string::String,
#[prost(message, repeated, tag = "3")]
pub column_filters: ::prost::alloc::vec::Vec<ColumnFilter>,
/// Optional fallback text search
#[prost(string, optional, tag = "4")]
pub text_query: ::core::option::Option<::prost::alloc::string::String>,
#[prost(int32, optional, tag = "5")]
pub limit: ::core::option::Option<i32>,
#[prost(string, optional, tag = "6")]
pub order_by: ::core::option::Option<::prost::alloc::string::String>,
#[prost(bool, optional, tag = "7")]
pub order_desc: ::core::option::Option<bool>,
}
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct ColumnFilter {
#[prost(string, tag = "1")]
pub column_name: ::prost::alloc::string::String,
#[prost(enumeration = "FilterType", tag = "2")]
pub filter_type: i32,
#[prost(string, tag = "3")]
pub value: ::prost::alloc::string::String,
/// For range queries
#[prost(string, optional, tag = "4")]
pub value2: ::core::option::Option<::prost::alloc::string::String>,
}
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct Search2Response {
#[prost(message, repeated, tag = "1")]
pub hits: ::prost::alloc::vec::Vec<search2_response::Hit>,
/// Total matching records (for pagination)
#[prost(int32, tag = "2")]
pub total_count: i32,
}
/// Nested message and enum types in `Search2Response`.
pub mod search2_response {
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct Hit {
#[prost(int64, tag = "1")]
pub id: i64,
/// No score - this is SQL-based
#[prost(string, tag = "2")]
pub content_json: ::prost::alloc::string::String,
/// Info about which columns matched
#[prost(string, optional, tag = "3")]
pub match_info: ::core::option::Option<::prost::alloc::string::String>,
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)]
#[repr(i32)]
pub enum FilterType {
Equals = 0,
Contains = 1,
StartsWith = 2,
EndsWith = 3,
Range = 4,
GreaterThan = 5,
LessThan = 6,
IsNull = 7,
IsNotNull = 8,
}
impl FilterType {
/// String value of the enum field names used in the ProtoBuf definition.
///
/// The values are not transformed in any way and thus are considered stable
/// (if the ProtoBuf definition does not change) and safe for programmatic use.
pub fn as_str_name(&self) -> &'static str {
match self {
Self::Equals => "EQUALS",
Self::Contains => "CONTAINS",
Self::StartsWith => "STARTS_WITH",
Self::EndsWith => "ENDS_WITH",
Self::Range => "RANGE",
Self::GreaterThan => "GREATER_THAN",
Self::LessThan => "LESS_THAN",
Self::IsNull => "IS_NULL",
Self::IsNotNull => "IS_NOT_NULL",
}
}
/// Creates an enum from field names used in the ProtoBuf definition.
pub fn from_str_name(value: &str) -> ::core::option::Option<Self> {
match value {
"EQUALS" => Some(Self::Equals),
"CONTAINS" => Some(Self::Contains),
"STARTS_WITH" => Some(Self::StartsWith),
"ENDS_WITH" => Some(Self::EndsWith),
"RANGE" => Some(Self::Range),
"GREATER_THAN" => Some(Self::GreaterThan),
"LESS_THAN" => Some(Self::LessThan),
"IS_NULL" => Some(Self::IsNull),
"IS_NOT_NULL" => Some(Self::IsNotNull),
_ => None,
}
}
}
/// Generated client implementations.
pub mod search2_client {
#![allow(
unused_variables,
dead_code,
missing_docs,
clippy::wildcard_imports,
clippy::let_unit_value,
)]
use tonic::codegen::*;
use tonic::codegen::http::Uri;
#[derive(Debug, Clone)]
pub struct Search2Client<T> {
inner: tonic::client::Grpc<T>,
}
impl Search2Client<tonic::transport::Channel> {
/// Attempt to create a new client by connecting to a given endpoint.
pub async fn connect<D>(dst: D) -> Result<Self, tonic::transport::Error>
where
D: TryInto<tonic::transport::Endpoint>,
D::Error: Into<StdError>,
{
let conn = tonic::transport::Endpoint::new(dst)?.connect().await?;
Ok(Self::new(conn))
}
}
impl<T> Search2Client<T>
where
T: tonic::client::GrpcService<tonic::body::Body>,
T::Error: Into<StdError>,
T::ResponseBody: Body<Data = Bytes> + std::marker::Send + 'static,
<T::ResponseBody as Body>::Error: Into<StdError> + std::marker::Send,
{
pub fn new(inner: T) -> Self {
let inner = tonic::client::Grpc::new(inner);
Self { inner }
}
pub fn with_origin(inner: T, origin: Uri) -> Self {
let inner = tonic::client::Grpc::with_origin(inner, origin);
Self { inner }
}
pub fn with_interceptor<F>(
inner: T,
interceptor: F,
) -> Search2Client<InterceptedService<T, F>>
where
F: tonic::service::Interceptor,
T::ResponseBody: Default,
T: tonic::codegen::Service<
http::Request<tonic::body::Body>,
Response = http::Response<
<T as tonic::client::GrpcService<tonic::body::Body>>::ResponseBody,
>,
>,
<T as tonic::codegen::Service<
http::Request<tonic::body::Body>,
>>::Error: Into<StdError> + std::marker::Send + std::marker::Sync,
{
Search2Client::new(InterceptedService::new(inner, interceptor))
}
/// Compress requests with the given encoding.
///
/// This requires the server to support it otherwise it might respond with an
/// error.
#[must_use]
pub fn send_compressed(mut self, encoding: CompressionEncoding) -> Self {
self.inner = self.inner.send_compressed(encoding);
self
}
/// Enable decompressing responses.
#[must_use]
pub fn accept_compressed(mut self, encoding: CompressionEncoding) -> Self {
self.inner = self.inner.accept_compressed(encoding);
self
}
/// Limits the maximum size of a decoded message.
///
/// Default: `4MB`
#[must_use]
pub fn max_decoding_message_size(mut self, limit: usize) -> Self {
self.inner = self.inner.max_decoding_message_size(limit);
self
}
/// Limits the maximum size of an encoded message.
///
/// Default: `usize::MAX`
#[must_use]
pub fn max_encoding_message_size(mut self, limit: usize) -> Self {
self.inner = self.inner.max_encoding_message_size(limit);
self
}
pub async fn search_table(
&mut self,
request: impl tonic::IntoRequest<super::Search2Request>,
) -> std::result::Result<
tonic::Response<super::Search2Response>,
tonic::Status,
> {
self.inner
.ready()
.await
.map_err(|e| {
tonic::Status::unknown(
format!("Service was not ready: {}", e.into()),
)
})?;
let codec = tonic::codec::ProstCodec::default();
let path = http::uri::PathAndQuery::from_static(
"/komp_ac.search2.Search2/SearchTable",
);
let mut req = request.into_request();
req.extensions_mut()
.insert(GrpcMethod::new("komp_ac.search2.Search2", "SearchTable"));
self.inner.unary(req, path, codec).await
}
}
}
/// Generated server implementations.
pub mod search2_server {
#![allow(
unused_variables,
dead_code,
missing_docs,
clippy::wildcard_imports,
clippy::let_unit_value,
)]
use tonic::codegen::*;
/// Generated trait containing gRPC methods that should be implemented for use with Search2Server.
#[async_trait]
pub trait Search2: std::marker::Send + std::marker::Sync + 'static {
async fn search_table(
&self,
request: tonic::Request<super::Search2Request>,
) -> std::result::Result<tonic::Response<super::Search2Response>, tonic::Status>;
}
#[derive(Debug)]
pub struct Search2Server<T> {
inner: Arc<T>,
accept_compression_encodings: EnabledCompressionEncodings,
send_compression_encodings: EnabledCompressionEncodings,
max_decoding_message_size: Option<usize>,
max_encoding_message_size: Option<usize>,
}
impl<T> Search2Server<T> {
pub fn new(inner: T) -> Self {
Self::from_arc(Arc::new(inner))
}
pub fn from_arc(inner: Arc<T>) -> Self {
Self {
inner,
accept_compression_encodings: Default::default(),
send_compression_encodings: Default::default(),
max_decoding_message_size: None,
max_encoding_message_size: None,
}
}
pub fn with_interceptor<F>(
inner: T,
interceptor: F,
) -> InterceptedService<Self, F>
where
F: tonic::service::Interceptor,
{
InterceptedService::new(Self::new(inner), interceptor)
}
/// Enable decompressing requests with the given encoding.
#[must_use]
pub fn accept_compressed(mut self, encoding: CompressionEncoding) -> Self {
self.accept_compression_encodings.enable(encoding);
self
}
/// Compress responses with the given encoding, if the client supports it.
#[must_use]
pub fn send_compressed(mut self, encoding: CompressionEncoding) -> Self {
self.send_compression_encodings.enable(encoding);
self
}
/// Limits the maximum size of a decoded message.
///
/// Default: `4MB`
#[must_use]
pub fn max_decoding_message_size(mut self, limit: usize) -> Self {
self.max_decoding_message_size = Some(limit);
self
}
/// Limits the maximum size of an encoded message.
///
/// Default: `usize::MAX`
#[must_use]
pub fn max_encoding_message_size(mut self, limit: usize) -> Self {
self.max_encoding_message_size = Some(limit);
self
}
}
impl<T, B> tonic::codegen::Service<http::Request<B>> for Search2Server<T>
where
T: Search2,
B: Body + std::marker::Send + 'static,
B::Error: Into<StdError> + std::marker::Send + 'static,
{
type Response = http::Response<tonic::body::Body>;
type Error = std::convert::Infallible;
type Future = BoxFuture<Self::Response, Self::Error>;
fn poll_ready(
&mut self,
_cx: &mut Context<'_>,
) -> Poll<std::result::Result<(), Self::Error>> {
Poll::Ready(Ok(()))
}
fn call(&mut self, req: http::Request<B>) -> Self::Future {
match req.uri().path() {
"/komp_ac.search2.Search2/SearchTable" => {
#[allow(non_camel_case_types)]
struct SearchTableSvc<T: Search2>(pub Arc<T>);
impl<T: Search2> tonic::server::UnaryService<super::Search2Request>
for SearchTableSvc<T> {
type Response = super::Search2Response;
type Future = BoxFuture<
tonic::Response<Self::Response>,
tonic::Status,
>;
fn call(
&mut self,
request: tonic::Request<super::Search2Request>,
) -> Self::Future {
let inner = Arc::clone(&self.0);
let fut = async move {
<T as Search2>::search_table(&inner, request).await
};
Box::pin(fut)
}
}
let accept_compression_encodings = self.accept_compression_encodings;
let send_compression_encodings = self.send_compression_encodings;
let max_decoding_message_size = self.max_decoding_message_size;
let max_encoding_message_size = self.max_encoding_message_size;
let inner = self.inner.clone();
let fut = async move {
let method = SearchTableSvc(inner);
let codec = tonic::codec::ProstCodec::default();
let mut grpc = tonic::server::Grpc::new(codec)
.apply_compression_config(
accept_compression_encodings,
send_compression_encodings,
)
.apply_max_message_size_config(
max_decoding_message_size,
max_encoding_message_size,
);
let res = grpc.unary(method, req).await;
Ok(res)
};
Box::pin(fut)
}
_ => {
Box::pin(async move {
let mut response = http::Response::new(
tonic::body::Body::default(),
);
let headers = response.headers_mut();
headers
.insert(
tonic::Status::GRPC_STATUS,
(tonic::Code::Unimplemented as i32).into(),
);
headers
.insert(
http::header::CONTENT_TYPE,
tonic::metadata::GRPC_CONTENT_TYPE,
);
Ok(response)
})
}
}
}
}
impl<T> Clone for Search2Server<T> {
fn clone(&self) -> Self {
let inner = self.inner.clone();
Self {
inner,
accept_compression_encodings: self.accept_compression_encodings,
send_compression_encodings: self.send_compression_encodings,
max_decoding_message_size: self.max_decoding_message_size,
max_encoding_message_size: self.max_encoding_message_size,
}
}
}
/// Generated gRPC service name
pub const SERVICE_NAME: &str = "komp_ac.search2.Search2";
impl<T> tonic::server::NamedService for Search2Server<T> {
const NAME: &'static str = SERVICE_NAME;
}
}

View File

@@ -10,14 +10,16 @@ use crate::server::services::{
TableDefinitionService,
TablesDataService,
TableScriptService,
AuthServiceImpl
AuthServiceImpl,
Search2Service,
};
use common::proto::komp_ac::{
table_structure::table_structure_service_server::TableStructureServiceServer,
table_definition::table_definition_server::TableDefinitionServer,
tables_data::tables_data_server::TablesDataServer,
table_script::table_script_server::TableScriptServer,
auth::auth_service_server::AuthServiceServer
auth::auth_service_server::AuthServiceServer,
search2::search2_server::Search2Server,
};
use search::{SearcherService, SearcherServer};
@@ -47,9 +49,8 @@ pub async fn run_server(db_pool: sqlx::PgPool) -> Result<(), Box<dyn std::error:
};
let table_script_service = TableScriptService { db_pool: db_pool.clone() };
let auth_service = AuthServiceImpl { db_pool: db_pool.clone() };
// MODIFIED: Instantiate SearcherService with the database pool
let search_service = SearcherService { pool: db_pool.clone() };
let search2_service = Search2Service { db_pool: db_pool.clone() };
Server::builder()
.add_service(TableStructureServiceServer::new(TableStructureHandler { db_pool: db_pool.clone() }))
@@ -58,6 +59,7 @@ pub async fn run_server(db_pool: sqlx::PgPool) -> Result<(), Box<dyn std::error:
.add_service(TableScriptServer::new(table_script_service))
.add_service(AuthServiceServer::new(auth_service))
.add_service(SearcherServer::new(search_service))
.add_service(Search2Server::new(search2_service))
.add_service(reflection_service)
.serve(addr)
.await?;

View File

@@ -5,9 +5,11 @@ pub mod table_definition_service;
pub mod tables_data_service;
pub mod table_script_service;
pub mod auth_service;
pub mod search2_service;
pub use table_structure_service::TableStructureHandler;
pub use table_definition_service::TableDefinitionService;
pub use tables_data_service::TablesDataService;
pub use table_script_service::TableScriptService;
pub use auth_service::AuthServiceImpl;
pub use search2_service::*;

View File

@@ -0,0 +1,202 @@
// src/server/services/search2_service.rs
use tonic::{Request, Response, Status};
use sqlx::PgPool;
use sqlx::Row;
use common::proto::komp_ac::search2::{
search2_server::Search2,
Search2Request, Search2Response, ColumnFilter, FilterType,
search2_response::Hit,
};
use crate::shared::schema_qualifier::qualify_table_name_for_data;
pub struct Search2Service {
pub db_pool: PgPool,
}
#[tonic::async_trait]
impl Search2 for Search2Service {
async fn search_table(
&self,
request: Request<Search2Request>,
) -> Result<Response<Search2Response>, Status> {
let req = request.into_inner();
// Build SQL query - NOW PASS DB_POOL AND AWAIT
let (sql, bind_values, count_sql) = build_search_query(&self.db_pool, &req).await
.map_err(|e| Status::invalid_argument(format!("Query build error: {}", e)))?;
// Execute main query
let rows = execute_query(&self.db_pool, &sql, &bind_values).await
.map_err(|e| Status::internal(format!("Database error: {}", e)))?;
// Execute count query for pagination
let total_count = execute_count_query(&self.db_pool, &count_sql, &bind_values).await
.map_err(|e| Status::internal(format!("Count query error: {}", e)))?;
// Convert to response format
let hits = rows.into_iter().map(|row| {
Hit {
id: row.id,
content_json: row.content_json,
match_info: Some(format!("SQL search on table: {}", req.table_name)),
}
}).collect();
Ok(Response::new(Search2Response {
hits,
total_count,
}))
}
}
#[derive(Debug)]
struct QueryRow {
id: i64,
content_json: String,
}
// MAKE THIS FUNCTION ASYNC AND ADD DB_POOL PARAMETER
async fn build_search_query(
db_pool: &PgPool,
request: &Search2Request
) -> Result<(String, Vec<String>, String), String> {
// Since Search2Request doesn't have profile_name, use "default"
// You can add profile_name to the proto later if needed
let profile_name = "default";
// NOW AWAIT THE ASYNC FUNCTION CALL
let qualified_table = qualify_table_name_for_data(db_pool, profile_name, &request.table_name)
.await
.map_err(|e| format!("Invalid table name: {}", e))?;
let mut conditions = vec!["deleted = FALSE".to_string()];
let mut bind_values = Vec::new();
// Build WHERE conditions from column filters
for filter in &request.column_filters {
let condition = build_filter_condition(filter, &mut bind_values)?;
conditions.push(condition);
}
// Add text search fallback if provided
if let Some(text_query) = &request.text_query {
if !text_query.trim().is_empty() {
bind_values.push(format!("%{}%", text_query.to_lowercase()));
conditions.push(format!(
"EXISTS (SELECT 1 FROM jsonb_each_text(to_jsonb(t)) as kv(key, value) WHERE LOWER(value) LIKE ${})",
bind_values.len()
));
}
}
let where_clause = conditions.join(" AND ");
// Build ORDER BY clause
let order_clause = if let Some(order_by) = &request.order_by {
let direction = if request.order_desc.unwrap_or(false) { "DESC" } else { "ASC" };
format!("ORDER BY \"{}\" {}", order_by, direction)
} else {
"ORDER BY id DESC".to_string()
};
let limit_clause = format!("LIMIT {}", request.limit.unwrap_or(100));
// Main query
let sql = format!(
"SELECT id, to_jsonb(t) AS data FROM {} t WHERE {} {} {}",
qualified_table, where_clause, order_clause, limit_clause
);
// Count query (for pagination)
let count_sql = format!(
"SELECT COUNT(*) FROM {} t WHERE {}",
qualified_table, where_clause
);
Ok((sql, bind_values, count_sql))
}
fn build_filter_condition(filter: &ColumnFilter, bind_values: &mut Vec<String>) -> Result<String, String> {
// FIX DEPRECATED WARNING - USE TryFrom INSTEAD
let filter_type = FilterType::try_from(filter.filter_type)
.map_err(|_| "Invalid filter type".to_string())?;
let param_idx = bind_values.len() + 1;
let condition = match filter_type {
FilterType::Equals => {
bind_values.push(filter.value.clone());
format!("\"{}\" = ${}", filter.column_name, param_idx)
},
FilterType::Contains => {
bind_values.push(format!("%{}%", filter.value));
format!("\"{}\" ILIKE ${}", filter.column_name, param_idx)
},
FilterType::StartsWith => {
bind_values.push(format!("{}%", filter.value));
format!("\"{}\" ILIKE ${}", filter.column_name, param_idx)
},
FilterType::EndsWith => {
bind_values.push(format!("%{}", filter.value));
format!("\"{}\" ILIKE ${}", filter.column_name, param_idx)
},
FilterType::Range => {
bind_values.push(filter.value.clone());
bind_values.push(filter.value2.as_ref().unwrap_or(&"".to_string()).clone());
format!("\"{}\" BETWEEN ${} AND ${}", filter.column_name, param_idx, param_idx + 1)
},
FilterType::GreaterThan => {
bind_values.push(filter.value.clone());
format!("\"{}\" > ${}", filter.column_name, param_idx)
},
FilterType::LessThan => {
bind_values.push(filter.value.clone());
format!("\"{}\" < ${}", filter.column_name, param_idx)
},
FilterType::IsNull => {
format!("\"{}\" IS NULL", filter.column_name)
},
FilterType::IsNotNull => {
format!("\"{}\" IS NOT NULL", filter.column_name)
},
};
Ok(condition)
}
async fn execute_query(pool: &PgPool, sql: &str, bind_values: &[String]) -> Result<Vec<QueryRow>, sqlx::Error> {
let mut query = sqlx::query(sql);
// Bind all parameters
for value in bind_values {
query = query.bind(value);
}
let rows = query.fetch_all(pool).await?;
let results = rows.into_iter().map(|row| {
QueryRow {
id: row.try_get("id").unwrap_or(0),
content_json: row.try_get::<serde_json::Value, _>("data")
.unwrap_or(serde_json::Value::Null)
.to_string(),
}
}).collect();
Ok(results)
}
async fn execute_count_query(pool: &PgPool, sql: &str, bind_values: &[String]) -> Result<i32, sqlx::Error> {
let mut query = sqlx::query(sql);
// Bind all parameters
for value in bind_values {
query = query.bind(value);
}
let row = query.fetch_one(pool).await?;
let count: i64 = row.try_get(0).unwrap_or(0);
Ok(count as i32)
}