refactor(handlers): extract QueryParams + IngestParams to reduce complexity
query_handler refactored: - Extract QueryParams struct with validation - Extract SearchMethod enum - Extract build_search_response helper - Extract apply_rbac_filter helper - Extract execute_hybrid_search helper - Complexity: 14 → 6 ingest_handler helpers: - Extract IngestParams struct with validation - Extract IngestParamsError with responses - Extract IngestResponse builder New tests (18 total): - QueryParams validation (10 tests) - IngestParams validation (8 tests) Total tests: 688 (was 670)
This commit is contained in:
@@ -19,6 +19,7 @@ use crate::gateway_queue_adapter::GatewayQueueAdapter;
|
||||
use crate::queue_worker::{QueueWorker, QueueWorkerConfig};
|
||||
use crate::queue_adapter::QueueAdapter;
|
||||
use crate::rbac::{AccessGuard, Claims as RbacClaims, builtin_role_provider, ResourceMeta, ResourceType, Verb, Visibility};
|
||||
use crate::handlers::{QueryParams, QueryParamsError, SearchMethod, build_search_response};
|
||||
|
||||
/// Server state with database and workers
|
||||
pub struct AppState {
|
||||
@@ -833,192 +834,115 @@ pub async fn query_handler(
|
||||
query: web::Query<std::collections::HashMap<String, String>>,
|
||||
state: web::Data<AppState>,
|
||||
) -> HttpResponse {
|
||||
// Auth + capability check
|
||||
let (claims, token) = match validate_auth(&req, &state).await {
|
||||
Ok(c) => c,
|
||||
Err(e) => return e,
|
||||
};
|
||||
|
||||
// Check read capability
|
||||
if !has_capability(&claims, "memory:read") {
|
||||
return HttpResponse::Forbidden().json(json!({
|
||||
"error": "forbidden",
|
||||
"reason": "missing capability: memory:read"
|
||||
}));
|
||||
}
|
||||
|
||||
if let Err(e) = check_rate_limit(&claims, &state, "/memory/query") {
|
||||
return e;
|
||||
}
|
||||
|
||||
let project = match query.get("project") {
|
||||
Some(p) => p.clone(),
|
||||
None => {
|
||||
return HttpResponse::BadRequest().json(json!({"error": "missing project parameter"}))
|
||||
}
|
||||
// Parse + validate params
|
||||
let params = match QueryParams::from_query(&query) {
|
||||
Ok(p) => p,
|
||||
Err(e) => return e.to_response(),
|
||||
};
|
||||
|
||||
let question = match query.get("query") {
|
||||
Some(q) => q.clone(),
|
||||
None => {
|
||||
return HttpResponse::BadRequest().json(json!({"error": "missing query parameter"}))
|
||||
}
|
||||
};
|
||||
|
||||
let limit = query
|
||||
.get("limit")
|
||||
.and_then(|l| l.parse::<i64>().ok())
|
||||
.unwrap_or(10);
|
||||
|
||||
let search_method = query.get("method").map(|s| s.as_str()).unwrap_or("hybrid");
|
||||
|
||||
// Get semantic results from pgvector (always run)
|
||||
let mut semantic_results = match state.query_worker.query(&project, &question, Some(50)).await {
|
||||
Ok(results) => results,
|
||||
// Execute semantic search
|
||||
let mut results = match state.query_worker.query(¶ms.project, ¶ms.question, Some(50)).await {
|
||||
Ok(r) => r,
|
||||
Err(e) => {
|
||||
tracing::error!("Semantic search failed: {}", e);
|
||||
return HttpResponse::InternalServerError().json(json!({"error": "semantic_search_failed"}));
|
||||
}
|
||||
};
|
||||
|
||||
// M3.8: Optimize search results if optimizer is available
|
||||
semantic_results = optimize_search_results(semantic_results, state.optimizer_service.as_ref()).await;
|
||||
// M3.8: Optimize results
|
||||
results = optimize_search_results(results, state.optimizer_service.as_ref()).await;
|
||||
|
||||
// RBAC: Filter results by access control
|
||||
if let Some(guard) = &state.access_guard {
|
||||
let rbac_claims = to_rbac_claims(&claims);
|
||||
let resources: Vec<ResourceMeta> = semantic_results
|
||||
.iter()
|
||||
.map(|r| query_result_to_resource_meta(r, &project))
|
||||
.collect();
|
||||
|
||||
let decisions = guard.check_access_batch(&rbac_claims, &resources, Verb::Read).await;
|
||||
|
||||
// Keep only allowed results
|
||||
semantic_results = semantic_results
|
||||
.into_iter()
|
||||
.zip(decisions.iter())
|
||||
.filter(|(_, decision)| decision.is_allowed())
|
||||
.map(|(result, _)| result)
|
||||
.collect();
|
||||
|
||||
tracing::debug!(
|
||||
"RBAC filtered {} results for user {}",
|
||||
decisions.iter().filter(|d| d.is_denied()).count(),
|
||||
claims.sub
|
||||
);
|
||||
// RBAC: Filter by access control
|
||||
results = apply_rbac_filter(&state, &claims, results, ¶ms.project).await;
|
||||
|
||||
// Route by search method
|
||||
match params.method {
|
||||
SearchMethod::Semantic => build_search_response(¶ms, results, None),
|
||||
SearchMethod::Hybrid => execute_hybrid_search(&state, ¶ms, results, &token).await,
|
||||
}
|
||||
}
|
||||
|
||||
// Handle different search methods
|
||||
match search_method {
|
||||
"semantic" => {
|
||||
// Return only semantic results
|
||||
let top_results = semantic_results
|
||||
.into_iter()
|
||||
.take(limit as usize)
|
||||
.collect::<Vec<_>>();
|
||||
|
||||
HttpResponse::Ok().json(json!({
|
||||
"query": question,
|
||||
"project": project,
|
||||
"method": "semantic",
|
||||
"results": top_results
|
||||
}))
|
||||
}
|
||||
"hybrid" => {
|
||||
// If OpenSearch is available, run hybrid search
|
||||
if let Some(os_client) = &state.opensearch_client {
|
||||
// Convert semantic results to format expected by hybrid_search
|
||||
let sem_results: Vec<(String, f32, String, String, Vec<String>)> = semantic_results
|
||||
.iter()
|
||||
.enumerate()
|
||||
.map(|(i, r)| {
|
||||
(
|
||||
format!("sem-{}", i),
|
||||
r.score,
|
||||
r.text.clone(),
|
||||
r.source.clone().unwrap_or_default(),
|
||||
r.provenance.clone(),
|
||||
)
|
||||
})
|
||||
.collect();
|
||||
/// Apply RBAC filtering to search results
|
||||
async fn apply_rbac_filter(
|
||||
state: &web::Data<AppState>,
|
||||
claims: &JwtClaims,
|
||||
results: Vec<crate::query_worker::QueryResult>,
|
||||
project: &str,
|
||||
) -> Vec<crate::query_worker::QueryResult> {
|
||||
let Some(guard) = &state.access_guard else {
|
||||
return results;
|
||||
};
|
||||
|
||||
let rbac_claims = to_rbac_claims(claims);
|
||||
let resources: Vec<ResourceMeta> = results
|
||||
.iter()
|
||||
.map(|r| query_result_to_resource_meta(r, project))
|
||||
.collect();
|
||||
|
||||
let decisions = guard.check_access_batch(&rbac_claims, &resources, Verb::Read).await;
|
||||
|
||||
let filtered: Vec<_> = results
|
||||
.into_iter()
|
||||
.zip(decisions.iter())
|
||||
.filter(|(_, d)| d.is_allowed())
|
||||
.map(|(r, _)| r)
|
||||
.collect();
|
||||
|
||||
tracing::debug!(
|
||||
"RBAC filtered {} results for user {}",
|
||||
decisions.iter().filter(|d| d.is_denied()).count(),
|
||||
claims.sub
|
||||
);
|
||||
|
||||
filtered
|
||||
}
|
||||
|
||||
let weights = HybridWeights {
|
||||
semantic: 0.6,
|
||||
lexical: 0.4,
|
||||
};
|
||||
/// Execute hybrid search with OpenSearch fallback
|
||||
async fn execute_hybrid_search(
|
||||
state: &web::Data<AppState>,
|
||||
params: &QueryParams,
|
||||
results: Vec<crate::query_worker::QueryResult>,
|
||||
token: &str,
|
||||
) -> HttpResponse {
|
||||
let Some(os_client) = &state.opensearch_client else {
|
||||
tracing::info!("OpenSearch not configured, using semantic search only");
|
||||
return build_search_response(params, results, Some("semantic_only"));
|
||||
};
|
||||
|
||||
match os_client.hybrid_search(&question, sem_results, &token, limit as usize, &weights).await {
|
||||
Ok(hybrid_results) => {
|
||||
// Serialize to include all result details
|
||||
let result_json: Vec<_> = semantic_results
|
||||
.into_iter()
|
||||
.take(limit as usize)
|
||||
.map(|r| json!({
|
||||
"level": r.level,
|
||||
"score": r.score,
|
||||
"text": r.text,
|
||||
"source": r.source,
|
||||
"provenance": r.provenance
|
||||
}))
|
||||
.collect();
|
||||
|
||||
HttpResponse::Ok().json(json!({
|
||||
"query": question,
|
||||
"project": project,
|
||||
"method": "hybrid",
|
||||
"results": result_json
|
||||
}))
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!("Hybrid search failed, falling back to semantic: {}", e);
|
||||
let result_json: Vec<_> = semantic_results
|
||||
.into_iter()
|
||||
.take(limit as usize)
|
||||
.map(|r| json!({
|
||||
"level": r.level,
|
||||
"score": r.score,
|
||||
"text": r.text,
|
||||
"source": r.source,
|
||||
"provenance": r.provenance
|
||||
}))
|
||||
.collect();
|
||||
|
||||
HttpResponse::Ok().json(json!({
|
||||
"query": question,
|
||||
"project": project,
|
||||
"method": "semantic_fallback",
|
||||
"results": result_json
|
||||
}))
|
||||
}
|
||||
}
|
||||
} else {
|
||||
// OpenSearch not available, fall back to semantic
|
||||
tracing::info!("OpenSearch not configured, using semantic search only");
|
||||
let result_json: Vec<_> = semantic_results
|
||||
.into_iter()
|
||||
.take(limit as usize)
|
||||
.map(|r| json!({
|
||||
"level": r.level,
|
||||
"score": r.score,
|
||||
"text": r.text,
|
||||
"source": r.source,
|
||||
"provenance": r.provenance
|
||||
}))
|
||||
.collect();
|
||||
|
||||
HttpResponse::Ok().json(json!({
|
||||
"query": question,
|
||||
"project": project,
|
||||
"method": "semantic_only",
|
||||
"results": result_json
|
||||
}))
|
||||
}
|
||||
}
|
||||
_ => {
|
||||
HttpResponse::BadRequest().json(json!({
|
||||
"error": "invalid_search_method",
|
||||
"valid_methods": ["semantic", "hybrid"]
|
||||
}))
|
||||
let sem_results: Vec<(String, f32, String, String, Vec<String>)> = results
|
||||
.iter()
|
||||
.enumerate()
|
||||
.map(|(i, r)| (
|
||||
format!("sem-{}", i),
|
||||
r.score,
|
||||
r.text.clone(),
|
||||
r.source.clone().unwrap_or_default(),
|
||||
r.provenance.clone(),
|
||||
))
|
||||
.collect();
|
||||
|
||||
let weights = HybridWeights { semantic: 0.6, lexical: 0.4 };
|
||||
|
||||
match os_client.hybrid_search(¶ms.question, sem_results, token, params.limit as usize, &weights).await {
|
||||
Ok(_) => build_search_response(params, results, Some("hybrid")),
|
||||
Err(e) => {
|
||||
tracing::warn!("Hybrid search failed, falling back to semantic: {}", e);
|
||||
build_search_response(params, results, Some("semantic_fallback"))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user