From 43778f730f27e949262b7e968e7dc4ee48a12e2b Mon Sep 17 00:00:00 2001 From: rock Date: Thu, 3 Sep 2026 09:12:38 -0700 Subject: [PATCH] refactor(handlers): extract QueryParams + IngestParams to reduce complexity MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- crates/mem-cli/src/handlers/ingest.rs | 200 +++++++++++++++++++++ crates/mem-cli/src/handlers/mod.rs | 10 ++ crates/mem-cli/src/handlers/query.rs | 218 +++++++++++++++++++++++ crates/mem-cli/src/http_server.rs | 242 +++++++++----------------- crates/mem-cli/src/lib.rs | 1 + 5 files changed, 512 insertions(+), 159 deletions(-) create mode 100644 crates/mem-cli/src/handlers/ingest.rs create mode 100644 crates/mem-cli/src/handlers/mod.rs create mode 100644 crates/mem-cli/src/handlers/query.rs diff --git a/crates/mem-cli/src/handlers/ingest.rs b/crates/mem-cli/src/handlers/ingest.rs new file mode 100644 index 0000000..d7d5d93 --- /dev/null +++ b/crates/mem-cli/src/handlers/ingest.rs @@ -0,0 +1,200 @@ +/// Ingest Handler Helpers +/// +/// Extracted to reduce ingest_handler complexity. + +use actix_web::HttpResponse; +use serde::{Deserialize, Serialize}; +use serde_json::json; + +// ============================================================================ +// Ingest Parameters +// ============================================================================ + +/// Validated ingest request +#[derive(Debug, Clone)] +pub struct IngestParams { + pub project: String, + pub ingest_id: String, + pub source: String, + pub records: Vec, +} + +#[derive(Debug, Clone, Deserialize)] +pub struct IngestRecord { + pub text: String, + #[serde(default)] + pub metadata: Option, +} + +/// Raw ingest request from API +#[derive(Debug, Clone, Deserialize)] +pub struct IngestRequestBody { + pub project: Option, + pub ingest_id: Option, + pub source: Option, + pub records: Option>, +} + +impl IngestParams { + /// Parse and validate ingest request body + pub fn from_body(body: IngestRequestBody) -> Result { + let project = body.project + .filter(|p| !p.is_empty()) + .ok_or(IngestParamsError::MissingProject)?; + + let ingest_id = body.ingest_id + .filter(|id| !id.is_empty()) + .ok_or(IngestParamsError::MissingIngestId)?; + + let source = body.source + .unwrap_or_else(|| "api".to_string()); + + let records = body.records + .filter(|r| !r.is_empty()) + .ok_or(IngestParamsError::EmptyRecords)?; + + // Validate each record has non-empty text + for (i, record) in records.iter().enumerate() { + if record.text.trim().is_empty() { + return Err(IngestParamsError::EmptyRecordText(i)); + } + } + + Ok(Self { project, ingest_id, source, records }) + } +} + +/// Ingest parameter validation errors +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum IngestParamsError { + MissingProject, + MissingIngestId, + EmptyRecords, + EmptyRecordText(usize), +} + +impl IngestParamsError { + pub fn to_response(&self) -> HttpResponse { + let reason = match self { + Self::MissingProject => "missing required field: project".to_string(), + Self::MissingIngestId => "missing required field: ingest_id".to_string(), + Self::EmptyRecords => "records array is empty".to_string(), + Self::EmptyRecordText(i) => format!("record {} has empty text", i), + }; + HttpResponse::BadRequest().json(json!({ + "error": "bad_request", + "reason": reason + })) + } +} + +// ============================================================================ +// Ingest Response Builder +// ============================================================================ + +#[derive(Debug, Clone, Serialize)] +pub struct IngestResponse { + pub status: String, + pub ingest_id: String, + pub records_queued: usize, +} + +impl IngestResponse { + pub fn accepted(ingest_id: &str, count: usize) -> HttpResponse { + HttpResponse::Accepted().json(Self { + status: "accepted".to_string(), + ingest_id: ingest_id.to_string(), + records_queued: count, + }) + } + + pub fn duplicate(ingest_id: &str) -> HttpResponse { + HttpResponse::Ok().json(json!({ + "status": "duplicate", + "ingest_id": ingest_id, + "message": "ingest_id already processed" + })) + } +} + +// ============================================================================ +// Tests +// ============================================================================ + +#[cfg(test)] +mod tests { + use super::*; + + fn make_body(project: Option<&str>, ingest_id: Option<&str>, records: Option>) -> IngestRequestBody { + IngestRequestBody { + project: project.map(|s| s.to_string()), + ingest_id: ingest_id.map(|s| s.to_string()), + source: None, + records: records.map(|r| r.into_iter().map(|t| IngestRecord { + text: t.to_string(), + metadata: None + }).collect()), + } + } + + #[test] + fn test_ingest_params_valid() { + let body = make_body(Some("homelab"), Some("test-001"), Some(vec!["fact 1", "fact 2"])); + let params = IngestParams::from_body(body).unwrap(); + + assert_eq!(params.project, "homelab"); + assert_eq!(params.ingest_id, "test-001"); + assert_eq!(params.source, "api"); + assert_eq!(params.records.len(), 2); + } + + #[test] + fn test_ingest_params_missing_project() { + let body = make_body(None, Some("test-001"), Some(vec!["fact"])); + assert_eq!(IngestParams::from_body(body).unwrap_err(), IngestParamsError::MissingProject); + } + + #[test] + fn test_ingest_params_empty_project() { + let body = make_body(Some(""), Some("test-001"), Some(vec!["fact"])); + assert_eq!(IngestParams::from_body(body).unwrap_err(), IngestParamsError::MissingProject); + } + + #[test] + fn test_ingest_params_missing_ingest_id() { + let body = make_body(Some("homelab"), None, Some(vec!["fact"])); + assert_eq!(IngestParams::from_body(body).unwrap_err(), IngestParamsError::MissingIngestId); + } + + #[test] + fn test_ingest_params_empty_ingest_id() { + let body = make_body(Some("homelab"), Some(""), Some(vec!["fact"])); + assert_eq!(IngestParams::from_body(body).unwrap_err(), IngestParamsError::MissingIngestId); + } + + #[test] + fn test_ingest_params_no_records() { + let body = make_body(Some("homelab"), Some("test-001"), None); + assert_eq!(IngestParams::from_body(body).unwrap_err(), IngestParamsError::EmptyRecords); + } + + #[test] + fn test_ingest_params_empty_records() { + let body = make_body(Some("homelab"), Some("test-001"), Some(vec![])); + assert_eq!(IngestParams::from_body(body).unwrap_err(), IngestParamsError::EmptyRecords); + } + + #[test] + fn test_ingest_params_empty_record_text() { + let body = IngestRequestBody { + project: Some("homelab".to_string()), + ingest_id: Some("test-001".to_string()), + source: None, + records: Some(vec![ + IngestRecord { text: "valid".to_string(), metadata: None }, + IngestRecord { text: " ".to_string(), metadata: None }, + ]), + }; + assert_eq!(IngestParams::from_body(body).unwrap_err(), IngestParamsError::EmptyRecordText(1)); + } +} diff --git a/crates/mem-cli/src/handlers/mod.rs b/crates/mem-cli/src/handlers/mod.rs new file mode 100644 index 0000000..f416234 --- /dev/null +++ b/crates/mem-cli/src/handlers/mod.rs @@ -0,0 +1,10 @@ +/// HTTP Handler Helpers +/// +/// Extracted from http_server.rs to reduce complexity. +/// Each handler has its own params struct with validation. + +pub mod query; +pub mod ingest; + +pub use query::*; +pub use ingest::*; diff --git a/crates/mem-cli/src/handlers/query.rs b/crates/mem-cli/src/handlers/query.rs new file mode 100644 index 0000000..284e28d --- /dev/null +++ b/crates/mem-cli/src/handlers/query.rs @@ -0,0 +1,218 @@ +/// Query Handler Helpers +/// +/// Extracted to reduce query_handler complexity. + +use actix_web::HttpResponse; +use serde::Serialize; +use serde_json::json; +use std::collections::HashMap; + +use crate::query_worker::QueryResult; + +// ============================================================================ +// Query Parameters +// ============================================================================ + +/// Search method for query endpoint +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub enum SearchMethod { + Semantic, + #[default] + Hybrid, +} + +impl SearchMethod { + pub fn from_str(s: &str) -> Self { + match s { + "semantic" => Self::Semantic, + _ => Self::Hybrid, + } + } + + pub fn as_str(&self) -> &'static str { + match self { + Self::Semantic => "semantic", + Self::Hybrid => "hybrid", + } + } +} + +/// Validated query parameters +#[derive(Debug, Clone)] +pub struct QueryParams { + pub project: String, + pub question: String, + pub limit: i64, + pub method: SearchMethod, +} + +impl QueryParams { + /// Parse and validate query parameters + pub fn from_query(query: &HashMap) -> Result { + let project = query.get("project") + .filter(|p| !p.is_empty()) + .ok_or(QueryParamsError::MissingProject)? + .clone(); + + let question = query.get("query") + .filter(|q| !q.is_empty()) + .ok_or(QueryParamsError::MissingQuery)? + .clone(); + + let limit = query.get("limit") + .and_then(|l| l.parse::().ok()) + .unwrap_or(10) + .clamp(1, 100); + + let method = query.get("method") + .map(|m| SearchMethod::from_str(m)) + .unwrap_or_default(); + + Ok(Self { project, question, limit, method }) + } +} + +/// Query parameter validation errors +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum QueryParamsError { + MissingProject, + MissingQuery, +} + +impl QueryParamsError { + pub fn to_response(&self) -> HttpResponse { + let reason = match self { + Self::MissingProject => "missing project parameter", + Self::MissingQuery => "missing query parameter", + }; + HttpResponse::BadRequest().json(json!({"error": reason})) + } +} + +// ============================================================================ +// Search Response Builder +// ============================================================================ + +#[derive(Debug, Clone, Serialize)] +pub struct SearchResultItem { + pub level: String, + pub score: f32, + pub text: String, + pub source: Option, + pub provenance: Vec, +} + +impl From for SearchResultItem { + fn from(r: QueryResult) -> Self { + Self { + level: r.level, + score: r.score, + text: r.text, + source: r.source, + provenance: r.provenance, + } + } +} + +/// Build search response JSON +pub fn build_search_response( + params: &QueryParams, + results: Vec, + method_override: Option<&str>, +) -> HttpResponse { + let items: Vec = results + .into_iter() + .take(params.limit as usize) + .map(SearchResultItem::from) + .collect(); + + HttpResponse::Ok().json(json!({ + "query": params.question, + "project": params.project, + "method": method_override.unwrap_or(params.method.as_str()), + "results": items + })) +} + +// ============================================================================ +// Tests +// ============================================================================ + +#[cfg(test)] +mod tests { + use super::*; + + fn make_query(pairs: &[(&str, &str)]) -> HashMap { + pairs.iter().map(|(k, v)| (k.to_string(), v.to_string())).collect() + } + + #[test] + fn test_query_params_valid() { + let query = make_query(&[("project", "homelab"), ("query", "kubernetes")]); + let params = QueryParams::from_query(&query).unwrap(); + + assert_eq!(params.project, "homelab"); + assert_eq!(params.question, "kubernetes"); + assert_eq!(params.limit, 10); + assert_eq!(params.method, SearchMethod::Hybrid); + } + + #[test] + fn test_query_params_with_limit() { + let query = make_query(&[("project", "test"), ("query", "foo"), ("limit", "25")]); + let params = QueryParams::from_query(&query).unwrap(); + assert_eq!(params.limit, 25); + } + + #[test] + fn test_query_params_limit_clamped_max() { + let query = make_query(&[("project", "test"), ("query", "foo"), ("limit", "999")]); + let params = QueryParams::from_query(&query).unwrap(); + assert_eq!(params.limit, 100); + } + + #[test] + fn test_query_params_limit_clamped_min() { + let query = make_query(&[("project", "test"), ("query", "foo"), ("limit", "0")]); + let params = QueryParams::from_query(&query).unwrap(); + assert_eq!(params.limit, 1); + } + + #[test] + fn test_query_params_semantic_method() { + let query = make_query(&[("project", "test"), ("query", "foo"), ("method", "semantic")]); + let params = QueryParams::from_query(&query).unwrap(); + assert_eq!(params.method, SearchMethod::Semantic); + } + + #[test] + fn test_query_params_missing_project() { + let query = make_query(&[("query", "foo")]); + assert_eq!(QueryParams::from_query(&query).unwrap_err(), QueryParamsError::MissingProject); + } + + #[test] + fn test_query_params_empty_project() { + let query = make_query(&[("project", ""), ("query", "foo")]); + assert_eq!(QueryParams::from_query(&query).unwrap_err(), QueryParamsError::MissingProject); + } + + #[test] + fn test_query_params_missing_query() { + let query = make_query(&[("project", "homelab")]); + assert_eq!(QueryParams::from_query(&query).unwrap_err(), QueryParamsError::MissingQuery); + } + + #[test] + fn test_query_params_empty_query() { + let query = make_query(&[("project", "homelab"), ("query", "")]); + assert_eq!(QueryParams::from_query(&query).unwrap_err(), QueryParamsError::MissingQuery); + } + + #[test] + fn test_search_method_from_str() { + assert_eq!(SearchMethod::from_str("semantic"), SearchMethod::Semantic); + assert_eq!(SearchMethod::from_str("hybrid"), SearchMethod::Hybrid); + assert_eq!(SearchMethod::from_str("unknown"), SearchMethod::Hybrid); + } +} diff --git a/crates/mem-cli/src/http_server.rs b/crates/mem-cli/src/http_server.rs index 63e5bac..98c7338 100644 --- a/crates/mem-cli/src/http_server.rs +++ b/crates/mem-cli/src/http_server.rs @@ -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>, state: web::Data, ) -> 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::().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 = 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::>(); - - 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)> = 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, + claims: &JwtClaims, + results: Vec, + project: &str, +) -> Vec { + let Some(guard) = &state.access_guard else { + return results; + }; + + let rbac_claims = to_rbac_claims(claims); + let resources: Vec = 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, + params: &QueryParams, + results: Vec, + 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)> = 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")) } } } diff --git a/crates/mem-cli/src/lib.rs b/crates/mem-cli/src/lib.rs index 97e0907..4596d00 100644 --- a/crates/mem-cli/src/lib.rs +++ b/crates/mem-cli/src/lib.rs @@ -1,4 +1,5 @@ pub mod endpoints; +pub mod handlers; pub mod http_server; pub mod ingest_worker; pub mod query_worker;