find.rs (7987B)
1 use radroots_replica_db::ReplicaSql; 2 use radroots_sql_core::SqliteExecutor; 3 4 use crate::cli::global::FindQueryArgs; 5 use crate::runtime::RuntimeError; 6 use crate::runtime::config::RuntimeConfig; 7 use crate::runtime::hyf::{self, HyfQueryRewriteRequest, HyfRequestContext}; 8 use crate::runtime::sync::{ 9 RelayIngestScope, freshness_for_scope_from_executor, freshness_requires_refresh, 10 market_refresh, missing_freshness, 11 }; 12 use crate::view::runtime::{ 13 FindHyfView, FindPriceView, FindQuantityView, FindResultHyfView, FindResultProvenanceView, 14 FindResultView, FindView, MarketReadinessView, 15 }; 16 17 const FIND_SOURCE: &str = "local replica · local first"; 18 const FIND_HYF_SOURCE: &str = "hyf query_rewrite · local first"; 19 const FIND_HYF_QUERY_REWRITE_REQUEST_ID: &str = "cli-find-query-rewrite"; 20 21 #[derive(Debug, Clone)] 22 struct AppliedQueryRewrite { 23 rewritten_query: String, 24 query_terms: Vec<String>, 25 } 26 27 impl AppliedQueryRewrite { 28 fn to_find_view(&self) -> FindHyfView { 29 FindHyfView { 30 state: "query_rewrite_applied".to_owned(), 31 source: FIND_HYF_SOURCE.to_owned(), 32 rewritten_query: self.rewritten_query.clone(), 33 query_terms: self.query_terms.clone(), 34 } 35 } 36 37 fn to_result_view(&self) -> FindResultHyfView { 38 FindResultHyfView { 39 state: "query_rewrite_applied".to_owned(), 40 rewritten_query: self.rewritten_query.clone(), 41 query_terms: self.query_terms.clone(), 42 } 43 } 44 } 45 46 pub fn search(config: &RuntimeConfig, args: &FindQueryArgs) -> Result<FindView, RuntimeError> { 47 let query = args.query.join(" "); 48 if !config.local.replica_db_path.exists() { 49 return Ok(FindView { 50 state: "unconfigured".to_owned(), 51 source: FIND_SOURCE.to_owned(), 52 query, 53 count: 0, 54 relay_count: config.relay.urls.len(), 55 replica_db: config.local.replica_db_path.display().to_string(), 56 freshness: missing_freshness(), 57 results: Vec::new(), 58 hyf: None, 59 reason: Some("local replica database is not initialized".to_owned()), 60 actions: vec!["radroots store init".to_owned()], 61 }); 62 } 63 64 refresh_market_if_needed(config)?; 65 let db = ReplicaSql::new(SqliteExecutor::open(&config.local.replica_db_path)?); 66 let freshness = 67 freshness_for_scope_from_executor(config, db.executor(), RelayIngestScope::MarketRefresh)?; 68 let applied_query_rewrite = attempt_query_rewrite(config, query.as_str(), &args.query); 69 let effective_query_terms = applied_query_rewrite 70 .as_ref() 71 .map(|rewrite| rewrite.query_terms.clone()) 72 .unwrap_or_else(|| normalize_query_terms(args.query.clone())); 73 let rows = db.trade_product_search(effective_query_terms.as_slice())?; 74 let relay_count = config.relay.urls.len(); 75 let result_provenance = FindResultProvenanceView { 76 origin: "local_replica.trade_product".to_owned(), 77 freshness: freshness.display.clone(), 78 relay_count, 79 }; 80 let results = rows 81 .into_iter() 82 .map(|row| { 83 let listing_addr = row.listing_addr.and_then(non_empty); 84 let primary_bin_id = row.primary_bin_id.and_then(non_empty); 85 let verified_primary_bin_id = row.verified_primary_bin_id.and_then(non_empty); 86 let available_amount = row.qty_avail; 87 let price_amount = row.price_amt; 88 let price_currency = row.price_currency; 89 let price_per_amount = row.price_qty_amt; 90 let readiness = MarketReadinessView::from_market_projection( 91 listing_addr.as_deref(), 92 primary_bin_id.as_deref(), 93 verified_primary_bin_id.as_deref(), 94 Some(row.title.as_str()), 95 Some(row.category.as_str()), 96 available_amount, 97 price_amount, 98 price_currency.as_str(), 99 price_per_amount, 100 ); 101 FindResultView { 102 id: row.id, 103 product_key: row.key, 104 readiness, 105 listing_addr, 106 primary_bin_id, 107 title: row.title, 108 category: row.category, 109 summary: non_empty(row.summary), 110 location_primary: row.location_primary.and_then(non_empty), 111 available: FindQuantityView { 112 total_amount: row.qty_amt, 113 total_unit: row.qty_unit, 114 label: row.qty_label.and_then(non_empty), 115 available_amount, 116 }, 117 price: FindPriceView { 118 amount: price_amount, 119 currency: price_currency, 120 per_amount: price_per_amount, 121 per_unit: row.price_qty_unit, 122 }, 123 provenance: result_provenance.clone(), 124 hyf: applied_query_rewrite 125 .as_ref() 126 .map(AppliedQueryRewrite::to_result_view), 127 } 128 }) 129 .collect::<Vec<_>>(); 130 131 let (state, reason, actions) = if results.is_empty() { 132 let actions = if freshness.state == "never" { 133 vec!["radroots sync status get".to_owned()] 134 } else { 135 Vec::new() 136 }; 137 ( 138 "empty".to_owned(), 139 Some(format!("no local market results matched `{query}`")), 140 actions, 141 ) 142 } else { 143 ("ready".to_owned(), None, Vec::new()) 144 }; 145 146 Ok(FindView { 147 state, 148 source: FIND_SOURCE.to_owned(), 149 query, 150 count: results.len(), 151 relay_count, 152 replica_db: config.local.replica_db_path.display().to_string(), 153 freshness, 154 results, 155 hyf: applied_query_rewrite.map(|rewrite| rewrite.to_find_view()), 156 reason, 157 actions, 158 }) 159 } 160 161 fn refresh_market_if_needed(config: &RuntimeConfig) -> Result<(), RuntimeError> { 162 if config.output.dry_run || config.relay.urls.is_empty() { 163 return Ok(()); 164 } 165 let executor = SqliteExecutor::open(&config.local.replica_db_path)?; 166 let freshness = 167 freshness_for_scope_from_executor(config, &executor, RelayIngestScope::MarketRefresh)?; 168 if freshness_requires_refresh(&freshness) { 169 let _ = market_refresh(config)?; 170 } 171 Ok(()) 172 } 173 174 fn attempt_query_rewrite( 175 config: &RuntimeConfig, 176 query: &str, 177 original_terms: &[String], 178 ) -> Option<AppliedQueryRewrite> { 179 if query.trim().is_empty() { 180 return None; 181 } 182 183 let client = hyf::resolve_runtime_client(config).ok()?; 184 let response = client 185 .query_rewrite( 186 FIND_HYF_QUERY_REWRITE_REQUEST_ID, 187 Some(FIND_HYF_QUERY_REWRITE_REQUEST_ID), 188 &HyfRequestContext::deterministic_cli(), 189 &HyfQueryRewriteRequest::new(query), 190 ) 191 .ok()?; 192 193 let rewritten_terms = normalize_query_terms(response.output.query_terms.clone()); 194 if rewritten_terms.is_empty() { 195 return None; 196 } 197 198 if rewritten_terms == normalize_query_terms(original_terms.iter().cloned()) { 199 return None; 200 } 201 202 let rewritten_query = { 203 let rewritten_text = response.output.rewritten_text.trim(); 204 if rewritten_text.is_empty() { 205 rewritten_terms.join(" ") 206 } else { 207 rewritten_text.to_owned() 208 } 209 }; 210 211 Some(AppliedQueryRewrite { 212 rewritten_query, 213 query_terms: rewritten_terms, 214 }) 215 } 216 217 fn non_empty(value: String) -> Option<String> { 218 let trimmed = value.trim(); 219 if trimmed.is_empty() { 220 None 221 } else { 222 Some(trimmed.to_owned()) 223 } 224 } 225 226 fn normalize_query_terms<I>(terms: I) -> Vec<String> 227 where 228 I: IntoIterator<Item = String>, 229 { 230 terms 231 .into_iter() 232 .map(|term| term.trim().to_lowercase()) 233 .filter(|term| !term.is_empty()) 234 .collect() 235 }