Sign in

flagon-io/g1t

public

Where people and agents ship software together. The open-source git platform for the whole job: issues, agents, checks and deploys to the edge.

gitagentscloudflarerustforgemcp

g1t/services/search/src/read.rs

490 lines19,515 bytesCodeBlame
1//! Answering searches, suggestions as someone types, and Explore.
2
3use std::collections::{HashMap, HashSet};
4
5use g1t_contracts::repos::{ReadableArgs, Repo};
6use g1t_contracts::search::*;
7use g1t_contracts::{Outcome, Viewer};
8use serde::Deserialize;
9use worker::Result;
10
11use crate::Search;
12use crate::query::{ItemKind, Query};
13use crate::rules::normalize_language;
14use crate::snippet;
15use crate::sql::{self, CodePlan, Page, Param, Sql};
16use crate::store;
17use crate::visibility::{Indexed, Reader, check};
18
19/// Results a page can reach: no further than the counts go.
20const MAX_PAGE: u32 = 50;
21/// Results of each kind in a suggestion.
22const SUGGEST_EACH: u32 = 4;
23/// Repositories on a page of Explore.
24const EXPLORE_PAGE: u32 = 24;
25/// Pieces of code read for each file on a page, at most.
26const PIECES_PER_FILE: u32 = 4;
27
28/// A path's segments made safe to put in an address.
29pub fn encode_path(path: &str) -> String {
30 path.split('/')
31 .map(|segment| {
32 let mut out = String::with_capacity(segment.len());
33 for byte in segment.bytes() {
34 match byte {
35 b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'.' | b'_' | b'~' | b'@' | b'+' | b'=' | b',' => {
36 out.push(byte as char)
37 }
38 _ => out.push_str(&format!("%{byte:02X}")),
39 }
40 }
41 out
42 })
43 .collect::<Vec<_>>()
44 .join("/")
45}
46
47fn split_labels(labels: &str) -> Vec<String> {
48 labels.split('|').filter(|label| !label.is_empty()).map(str::to_owned).collect()
49}
50
51fn split_topics(topics: &str) -> Vec<String> {
52 topics.split_whitespace().map(str::to_owned).collect()
53}
54
55#[derive(Deserialize)]
56struct RepoRow {
57 repo_id: String,
58 namespace: String,
59 name: String,
60 description: Option<String>,
61 topics: String,
62 language: Option<String>,
63 private: u8,
64 pushed_at: Option<String>,
65 created_at: String,
66 #[serde(default)]
67 readme: Option<String>,
68}
69
70#[derive(Deserialize)]
71struct FileHit {
72 fid: i64,
73}
74
75#[derive(Deserialize)]
76struct FileRow {
77 fid: i64,
78 path: String,
79 language: Option<String>,
80 repo_id: String,
81 namespace: String,
82 name: String,
83 default_branch: String,
84 private: u8,
85}
86
87#[derive(Deserialize)]
88struct PieceRow {
89 fid: i64,
90 start_line: u32,
91 content: String,
92}
93
94#[derive(Deserialize)]
95struct ItemRow {
96 kind: String,
97 number: u32,
98 title: String,
99 state: String,
100 status: String,
101 author: String,
102 labels: String,
103 updated_at: String,
104 repo_id: String,
105 namespace: String,
106 name: String,
107 private: u8,
108 body: Option<String>,
109}
110
111#[derive(Deserialize)]
112struct PersonRow {
113 kind: String,
114 slug: String,
115 name: Option<String>,
116 bio: Option<String>,
117 avatar: Option<String>,
118}
119
120/// A result, and the repository whose visibility decides whether it is shown.
121struct Found {
122 hit: Hit,
123 repo: Option<Indexed>,
124}
125
126fn repo_hit(row: RepoRow, terms: &[String]) -> Found {
127 let full = format!("{}/{}", row.namespace, row.name);
128 let mut hit = Hit::new(HitKind::Repository, full.clone(), format!("/{}/{}", row.namespace, row.name));
129 hit.repo = Some(full);
130 hit.private = row.private != 0;
131 hit.snippet = match (&row.description, &row.readme) {
132 (Some(description), _) if !description.trim().is_empty() => snippet::prose(description, terms),
133 (_, Some(readme)) if snippet::contains_any(readme, terms) => snippet::prose(readme, terms),
134 _ => Vec::new(),
135 };
136 hit.description = row.description;
137 hit.language = row.language;
138 hit.topics = split_topics(&row.topics);
139 hit.updated_at = row.pushed_at.or(Some(row.created_at));
140 Found {
141 hit,
142 repo: Some(Indexed {
143 repo_id: row.repo_id,
144 namespace: row.namespace,
145 private: row.private != 0,
146 }),
147 }
148}
149
150fn item_hit(row: ItemRow, terms: &[String]) -> Found {
151 let pull = row.kind == "pull";
152 let full = format!("{}/{}", row.namespace, row.name);
153 let url = format!("/{full}/{}/{}", if pull { "pull" } else { "issues" }, row.number);
154 let mut hit = Hit::new(if pull { HitKind::Pull } else { HitKind::Issue }, row.title, url);
155 hit.repo = Some(full);
156 hit.private = row.private != 0;
157 hit.snippet = snippet::prose(row.body.as_deref().unwrap_or_default(), terms);
158 hit.number = Some(row.number);
159 hit.state = Some(if pull { row.status } else { row.state });
160 hit.author = Some(row.author);
161 hit.labels = split_labels(&row.labels);
162 hit.updated_at = Some(row.updated_at);
163 Found {
164 hit,
165 repo: Some(Indexed {
166 repo_id: row.repo_id,
167 namespace: row.namespace,
168 private: row.private != 0,
169 }),
170 }
171}
172
173fn person_hit(row: PersonRow, terms: &[String]) -> Found {
174 let workspace = row.kind == "workspace";
175 let url = if workspace { format!("/{}", row.slug) } else { format!("/u/{}", row.slug) };
176 let title = row.name.clone().filter(|name| !name.trim().is_empty()).unwrap_or_else(|| row.slug.clone());
177 let mut hit = Hit::new(if workspace { HitKind::Workspace } else { HitKind::User }, title, url);
178 hit.slug = Some(row.slug);
179 hit.snippet = row.bio.as_deref().map(|bio| snippet::prose(bio, terms)).unwrap_or_default();
180 hit.description = row.bio;
181 hit.avatar = row.avatar;
182 Found { hit, repo: None }
183}
184
185impl Search {
186 /// The second visibility check: of these repositories, the ones the
187 /// repos service says this viewer may read now, and of those, the ones
188 /// that are archived. Corrects the index where it was behind.
189 async fn readable(&self, viewer: &Viewer, indexed: &[Indexed]) -> (HashSet<String>, HashSet<String>) {
190 if indexed.is_empty() {
191 return (HashSet::new(), HashSet::new());
192 }
193 let ids: Vec<String> = indexed.iter().map(|row| row.repo_id.clone()).collect::<HashSet<_>>().into_iter().collect();
194 let readable: Result<Vec<Repo>> =
195 g1t_kit::call(&self.repos, "readable", &ReadableArgs { ids, viewer: viewer.clone() }).await;
196 if let Err(error) = &readable {
197 worker::console_error!("search: could not check visibility: {error}");
198 }
199 let verdict = check(&Reader::of(viewer), indexed, readable.as_deref().ok());
200 let archived: HashSet<String> = readable
201 .as_deref()
202 .unwrap_or_default()
203 .iter()
204 .filter(|repo| repo.archived())
205 .map(|repo| repo.id.clone())
206 .collect();
207 if !verdict.corrections.is_empty() {
208 let mut statements = Vec::new();
209 for correction in &verdict.corrections {
210 let (text, params) = match &correction.path {
211 Some((namespace, name)) => (
212 "UPDATE repos SET private = ?, namespace = ?, name = ? WHERE repo_id = ?",
213 vec![
214 Param::Int(i64::from(correction.private)),
215 Param::Text(namespace.clone()),
216 Param::Text(name.clone()),
217 Param::Text(correction.repo_id.clone()),
218 ],
219 ),
220 None => (
221 "UPDATE repos SET private = ? WHERE repo_id = ?",
222 vec![Param::Int(i64::from(correction.private)), Param::Text(correction.repo_id.clone())],
223 ),
224 };
225 if let Ok(statement) = store::prepare(&self.db, text, params) {
226 statements.push(statement);
227 }
228 }
229 if let Err(error) = store::run_all(&self.db, statements).await {
230 worker::console_error!("search: could not correct visibility: {error}");
231 }
232 }
233 (verdict.keep, archived)
234 }
235
236 /// Results the viewer may see, in order.
237 async fn admit(&self, viewer: &Viewer, found: Vec<Found>) -> Vec<Hit> {
238 let indexed: Vec<Indexed> = found.iter().filter_map(|f| f.repo.clone()).collect();
239 let (keep, _) = self.readable(viewer, &indexed).await;
240 found
241 .into_iter()
242 .filter(|f| f.repo.as_ref().is_none_or(|repo| keep.contains(&repo.repo_id)))
243 .map(|f| f.hit)
244 .collect()
245 }
246
247 async fn count_of(&self, kind: SearchType, query: &Query, reader: &Reader) -> Result<u32> {
248 if !query.applies_to(kind) {
249 return Ok(0);
250 }
251 let sql = match kind {
252 SearchType::Repositories => Some(sql::repos_count(query, reader)),
253 SearchType::Code => sql::code_plan(query).ok().map(|plan| sql::code_count(&plan, query, reader)),
254 SearchType::Issues => Some(sql::items_count(query, reader, ItemKind::Issue)),
255 SearchType::Pulls => Some(sql::items_count(query, reader, ItemKind::Pull)),
256 SearchType::People => sql::people_select(query).map(|select| sql::count(&select)),
257 };
258 match sql {
259 Some(sql) => Ok(store::count(&self.db, &sql).await?.min(COUNT_CAP)),
260 None => Ok(0),
261 }
262 }
263
264 async fn code_hits(&self, plan: &CodePlan, query: &Query, reader: &Reader, page: Page, terms: &[String]) -> Result<Vec<Found>> {
265 let fids: Vec<i64> = store::all::<FileHit>(&self.db, &sql::code_page(plan, query, reader, page))
266 .await?
267 .into_iter()
268 .map(|row| row.fid)
269 .collect();
270 if fids.is_empty() {
271 return Ok(Vec::new());
272 }
273 let files = store::all::<FileRow>(&self.db, &sql::code_files(&fids)).await?;
274 let mut pieces: HashMap<i64, Vec<PieceRow>> = HashMap::new();
275 for piece in store::all::<PieceRow>(&self.db, &sql::code_pieces(plan, &fids, page.limit * PIECES_PER_FILE)).await? {
276 pieces.entry(piece.fid).or_default().push(piece);
277 }
278 // Files that matched by name have no matching piece: their first.
279 let missing: Vec<i64> = fids.iter().copied().filter(|fid| !pieces.contains_key(fid)).collect();
280 if !missing.is_empty() {
281 for piece in store::all::<PieceRow>(&self.db, &sql::first_pieces(&missing)).await? {
282 pieces.entry(piece.fid).or_default().push(piece);
283 }
284 }
285 let by_id: HashMap<i64, FileRow> = files.into_iter().map(|file| (file.fid, file)).collect();
286 let mut found = Vec::new();
287 for fid in fids {
288 let Some(file) = by_id.get(&fid) else { continue };
289 let parts: Vec<(u32, &str)> = pieces
290 .get(&fid)
291 .map(|list| list.iter().take(PIECES_PER_FILE as usize).map(|p| (p.start_line, p.content.as_str())).collect())
292 .unwrap_or_default();
293 let lines = snippet::code(&parts, terms);
294 let line = snippet::first_match(&lines);
295 let full = format!("{}/{}", file.namespace, file.name);
296 let url = format!(
297 "/{full}/blob/{}/{}{}",
298 encode_path(&file.default_branch),
299 encode_path(&file.path),
300 line.map(|n| format!("#L{n}")).unwrap_or_default()
301 );
302 let mut hit = Hit::new(HitKind::Code, file.path.clone(), url);
303 hit.repo = Some(full);
304 hit.private = file.private != 0;
305 hit.path = Some(file.path.clone());
306 hit.language = file.language.clone();
307 hit.git_ref = Some(file.default_branch.clone());
308 hit.lines = lines;
309 found.push(Found {
310 hit,
311 repo: Some(Indexed {
312 repo_id: file.repo_id.clone(),
313 namespace: file.namespace.clone(),
314 private: file.private != 0,
315 }),
316 });
317 }
318 Ok(found)
319 }
320
321 async fn page_of(&self, kind: SearchType, query: &Query, reader: &Reader, page: Page, terms: &[String]) -> Result<Vec<Found>> {
322 if !query.applies_to(kind) {
323 return Ok(Vec::new());
324 }
325 Ok(match kind {
326 SearchType::Repositories => store::all::<RepoRow>(&self.db, &sql::repos_page(query, reader, page))
327 .await?
328 .into_iter()
329 .map(|row| repo_hit(row, terms))
330 .collect(),
331 SearchType::Code => match sql::code_plan(query) {
332 Ok(plan) => self.code_hits(&plan, query, reader, page, terms).await?,
333 Err(_) => Vec::new(),
334 },
335 SearchType::Issues | SearchType::Pulls => {
336 let item = if kind == SearchType::Issues { ItemKind::Issue } else { ItemKind::Pull };
337 store::all::<ItemRow>(&self.db, &sql::items_page(query, reader, item, page))
338 .await?
339 .into_iter()
340 .map(|row| item_hit(row, terms))
341 .collect()
342 }
343 SearchType::People => match sql::people_page(query, page) {
344 Some(sql) => store::all::<PersonRow>(&self.db, &sql)
345 .await?
346 .into_iter()
347 .map(|row| person_hit(row, terms))
348 .collect(),
349 None => Vec::new(),
350 },
351 })
352 }
353
354 pub async fn search(&self, a: SearchArgs) -> Result<Outcome<SearchResults>> {
355 let query = Query::parse(&a.query);
356 let reader = Reader::of(&a.viewer);
357 let kind = a.kind.unwrap_or_else(|| query.default_type());
358 let per_page = a.per_page.unwrap_or(DEFAULT_PER_PAGE).clamp(1, MAX_PER_PAGE);
359 let page = a.page.unwrap_or(1).clamp(1, MAX_PAGE);
360 let mut results = SearchResults {
361 query: query.display(),
362 kind,
363 counts: SearchCounts::default(),
364 page,
365 per_page,
366 more: false,
367 hits: Vec::new(),
368 notes: query.notes.clone(),
369 };
370 if query.is_empty() {
371 results.notes.push("Type a word, a \"phrase\" or a qualifier such as repo:owner/name.".into());
372 return Ok(Outcome::Ok(results));
373 }
374 if kind == SearchType::Code
375 && let Err(why) = sql::code_plan(&query)
376 {
377 results.notes.push(why.to_owned());
378 }
379 for each in SearchType::ALL {
380 let counted = match self.count_of(each, &query, &reader).await {
381 Ok(count) => count,
382 Err(error) => {
383 worker::console_error!("search: counting {} failed: {error}", each.as_str());
384 0
385 }
386 };
387 results.counts.set(each, counted);
388 }
389 // One more than a page, to know whether there is another.
390 let window = Page { limit: per_page + 1, offset: (page - 1) * per_page };
391 let terms = query.highlights();
392 let found = match self.page_of(kind, &query, &reader, window, &terms).await {
393 Ok(found) => found,
394 Err(error) => {
395 // Most often a query the index could not read.
396 worker::console_error!("search: {} failed for {:?}: {error}", kind.as_str(), results.query);
397 results.notes.push("Part of that query could not be read. Try fewer symbols, or put them in quotes.".into());
398 Vec::new()
399 }
400 };
401 let before = found.len();
402 let mut hits = self.admit(&a.viewer, found).await;
403 // What the second check dropped is not counted either.
404 let dropped = (before - hits.len()) as u32;
405 results.counts.set(kind, results.counts.get(kind).saturating_sub(dropped));
406 results.more = hits.len() > per_page as usize;
407 hits.truncate(per_page as usize);
408 results.hits = hits;
409 Ok(Outcome::Ok(results))
410 }
411
412 /// A few repositories, issues, pull requests and people, for the
413 /// command palette. Never code: that is a full search away.
414 pub async fn suggest(&self, a: SuggestArgs) -> Result<Vec<Hit>> {
415 let query = Query::parse(&a.query);
416 if query.prose_match().is_none() {
417 return Ok(Vec::new());
418 }
419 let reader = Reader::of(&a.viewer);
420 let page = Page { limit: SUGGEST_EACH, offset: 0 };
421 let terms = query.highlights();
422 let mut found = Vec::new();
423 for kind in [SearchType::Repositories, SearchType::Issues, SearchType::Pulls, SearchType::People] {
424 match self.page_of(kind, &query, &reader, page, &terms).await {
425 Ok(more) => found.extend(more),
426 Err(error) => worker::console_error!("search: suggesting {} failed: {error}", kind.as_str()),
427 }
428 }
429 Ok(self.admit(&a.viewer, found).await)
430 }
431
432 pub async fn explore(&self, a: ExploreArgs) -> Result<Explore> {
433 let newest = a.sort.as_deref() == Some("new");
434 let language = a.language.as_deref().map(str::trim).filter(|l| !l.is_empty()).map(normalize_language);
435 let topic = a.topic.as_deref().map(|t| t.trim().to_lowercase()).filter(|t| !t.is_empty());
436 let page = a.page.unwrap_or(1).clamp(1, MAX_PAGE);
437 let window = Page { limit: EXPLORE_PAGE + 1, offset: (page - 1) * EXPLORE_PAGE };
438 let rows = store::all::<RepoRow>(&self.db, &sql::explore(newest, language.as_deref(), topic.as_deref(), window)).await?;
439 let indexed: Vec<Indexed> = rows
440 .iter()
441 .map(|row| Indexed { repo_id: row.repo_id.clone(), namespace: row.namespace.clone(), private: row.private != 0 })
442 .collect();
443 // Explore is for everyone: checked as if signed out, whoever asks.
444 let (keep, archived) = self.readable(&None, &indexed).await;
445 let mut repos: Vec<ExploreRepo> = rows
446 .into_iter()
447 .filter(|row| keep.contains(&row.repo_id))
448 .map(|row| ExploreRepo {
449 archived: archived.contains(&row.repo_id),
450 namespace: row.namespace,
451 name: row.name,
452 description: row.description,
453 topics: split_topics(&row.topics),
454 language: row.language,
455 created_at: row.created_at,
456 pushed_at: row.pushed_at,
457 })
458 .collect();
459 let more = repos.len() > EXPLORE_PAGE as usize;
460 repos.truncate(EXPLORE_PAGE as usize);
461 let (languages, topics) = if page == 1 {
462 (
463 store::all::<Facet>(&self.db, &Sql { text: sql::LANGUAGE_FACETS.into(), params: vec![] }).await?,
464 store::all::<Facet>(&self.db, &Sql { text: sql::TOPIC_FACETS.into(), params: vec![] }).await?,
465 )
466 } else {
467 (Vec::new(), Vec::new())
468 };
469 Ok(Explore { repos, languages, topics, page, more })
470 }
471}
472
473#[cfg(test)]
474mod tests {
475 use super::*;
476
477 #[test]
478 fn paths_are_safe_in_addresses() {
479 assert_eq!(encode_path("src/main.rs"), "src/main.rs");
480 assert_eq!(encode_path("docs/a b#c?.md"), "docs/a%20b%23c%3F.md");
481 assert_eq!(encode_path("src/ünï.rs"), "src/%C3%BCn%C3%AF.rs");
482 }
483
484 #[test]
485 fn labels_and_topics_are_split() {
486 assert_eq!(split_labels("|bug|good first issue|"), vec!["bug", "good first issue"]);
487 assert!(split_labels("|").is_empty());
488 assert_eq!(split_topics("cli rust"), vec!["cli", "rust"]);
489 }
490}