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

482 lines19,113 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. Corrects the index
188 /// where it was behind.
189 async fn readable(&self, viewer: &Viewer, indexed: &[Indexed]) -> HashSet<String> {
190 if indexed.is_empty() {
191 return 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 if !verdict.corrections.is_empty() {
201 let mut statements = Vec::new();
202 for correction in &verdict.corrections {
203 let (text, params) = match &correction.path {
204 Some((namespace, name)) => (
205 "UPDATE repos SET private = ?, namespace = ?, name = ? WHERE repo_id = ?",
206 vec![
207 Param::Int(i64::from(correction.private)),
208 Param::Text(namespace.clone()),
209 Param::Text(name.clone()),
210 Param::Text(correction.repo_id.clone()),
211 ],
212 ),
213 None => (
214 "UPDATE repos SET private = ? WHERE repo_id = ?",
215 vec![Param::Int(i64::from(correction.private)), Param::Text(correction.repo_id.clone())],
216 ),
217 };
218 if let Ok(statement) = store::prepare(&self.db, text, params) {
219 statements.push(statement);
220 }
221 }
222 if let Err(error) = store::run_all(&self.db, statements).await {
223 worker::console_error!("search: could not correct visibility: {error}");
224 }
225 }
226 verdict.keep
227 }
228
229 /// Results the viewer may see, in order.
230 async fn admit(&self, viewer: &Viewer, found: Vec<Found>) -> Vec<Hit> {
231 let indexed: Vec<Indexed> = found.iter().filter_map(|f| f.repo.clone()).collect();
232 let keep = self.readable(viewer, &indexed).await;
233 found
234 .into_iter()
235 .filter(|f| f.repo.as_ref().is_none_or(|repo| keep.contains(&repo.repo_id)))
236 .map(|f| f.hit)
237 .collect()
238 }
239
240 async fn count_of(&self, kind: SearchType, query: &Query, reader: &Reader) -> Result<u32> {
241 if !query.applies_to(kind) {
242 return Ok(0);
243 }
244 let sql = match kind {
245 SearchType::Repositories => Some(sql::repos_count(query, reader)),
246 SearchType::Code => sql::code_plan(query).ok().map(|plan| sql::code_count(&plan, query, reader)),
247 SearchType::Issues => Some(sql::items_count(query, reader, ItemKind::Issue)),
248 SearchType::Pulls => Some(sql::items_count(query, reader, ItemKind::Pull)),
249 SearchType::People => sql::people_select(query).map(|select| sql::count(&select)),
250 };
251 match sql {
252 Some(sql) => Ok(store::count(&self.db, &sql).await?.min(COUNT_CAP)),
253 None => Ok(0),
254 }
255 }
256
257 async fn code_hits(&self, plan: &CodePlan, query: &Query, reader: &Reader, page: Page, terms: &[String]) -> Result<Vec<Found>> {
258 let fids: Vec<i64> = store::all::<FileHit>(&self.db, &sql::code_page(plan, query, reader, page))
259 .await?
260 .into_iter()
261 .map(|row| row.fid)
262 .collect();
263 if fids.is_empty() {
264 return Ok(Vec::new());
265 }
266 let files = store::all::<FileRow>(&self.db, &sql::code_files(&fids)).await?;
267 let mut pieces: HashMap<i64, Vec<PieceRow>> = HashMap::new();
268 for piece in store::all::<PieceRow>(&self.db, &sql::code_pieces(plan, &fids, page.limit * PIECES_PER_FILE)).await? {
269 pieces.entry(piece.fid).or_default().push(piece);
270 }
271 // Files that matched by name have no matching piece: their first.
272 let missing: Vec<i64> = fids.iter().copied().filter(|fid| !pieces.contains_key(fid)).collect();
273 if !missing.is_empty() {
274 for piece in store::all::<PieceRow>(&self.db, &sql::first_pieces(&missing)).await? {
275 pieces.entry(piece.fid).or_default().push(piece);
276 }
277 }
278 let by_id: HashMap<i64, FileRow> = files.into_iter().map(|file| (file.fid, file)).collect();
279 let mut found = Vec::new();
280 for fid in fids {
281 let Some(file) = by_id.get(&fid) else { continue };
282 let parts: Vec<(u32, &str)> = pieces
283 .get(&fid)
284 .map(|list| list.iter().take(PIECES_PER_FILE as usize).map(|p| (p.start_line, p.content.as_str())).collect())
285 .unwrap_or_default();
286 let lines = snippet::code(&parts, terms);
287 let line = snippet::first_match(&lines);
288 let full = format!("{}/{}", file.namespace, file.name);
289 let url = format!(
290 "/{full}/blob/{}/{}{}",
291 encode_path(&file.default_branch),
292 encode_path(&file.path),
293 line.map(|n| format!("#L{n}")).unwrap_or_default()
294 );
295 let mut hit = Hit::new(HitKind::Code, file.path.clone(), url);
296 hit.repo = Some(full);
297 hit.private = file.private != 0;
298 hit.path = Some(file.path.clone());
299 hit.language = file.language.clone();
300 hit.git_ref = Some(file.default_branch.clone());
301 hit.lines = lines;
302 found.push(Found {
303 hit,
304 repo: Some(Indexed {
305 repo_id: file.repo_id.clone(),
306 namespace: file.namespace.clone(),
307 private: file.private != 0,
308 }),
309 });
310 }
311 Ok(found)
312 }
313
314 async fn page_of(&self, kind: SearchType, query: &Query, reader: &Reader, page: Page, terms: &[String]) -> Result<Vec<Found>> {
315 if !query.applies_to(kind) {
316 return Ok(Vec::new());
317 }
318 Ok(match kind {
319 SearchType::Repositories => store::all::<RepoRow>(&self.db, &sql::repos_page(query, reader, page))
320 .await?
321 .into_iter()
322 .map(|row| repo_hit(row, terms))
323 .collect(),
324 SearchType::Code => match sql::code_plan(query) {
325 Ok(plan) => self.code_hits(&plan, query, reader, page, terms).await?,
326 Err(_) => Vec::new(),
327 },
328 SearchType::Issues | SearchType::Pulls => {
329 let item = if kind == SearchType::Issues { ItemKind::Issue } else { ItemKind::Pull };
330 store::all::<ItemRow>(&self.db, &sql::items_page(query, reader, item, page))
331 .await?
332 .into_iter()
333 .map(|row| item_hit(row, terms))
334 .collect()
335 }
336 SearchType::People => match sql::people_page(query, page) {
337 Some(sql) => store::all::<PersonRow>(&self.db, &sql)
338 .await?
339 .into_iter()
340 .map(|row| person_hit(row, terms))
341 .collect(),
342 None => Vec::new(),
343 },
344 })
345 }
346
347 pub async fn search(&self, a: SearchArgs) -> Result<Outcome<SearchResults>> {
348 let query = Query::parse(&a.query);
349 let reader = Reader::of(&a.viewer);
350 let kind = a.kind.unwrap_or_else(|| query.default_type());
351 let per_page = a.per_page.unwrap_or(DEFAULT_PER_PAGE).clamp(1, MAX_PER_PAGE);
352 let page = a.page.unwrap_or(1).clamp(1, MAX_PAGE);
353 let mut results = SearchResults {
354 query: query.display(),
355 kind,
356 counts: SearchCounts::default(),
357 page,
358 per_page,
359 more: false,
360 hits: Vec::new(),
361 notes: query.notes.clone(),
362 };
363 if query.is_empty() {
364 results.notes.push("Type a word, a \"phrase\" or a qualifier such as repo:owner/name.".into());
365 return Ok(Outcome::Ok(results));
366 }
367 if kind == SearchType::Code
368 && let Err(why) = sql::code_plan(&query)
369 {
370 results.notes.push(why.to_owned());
371 }
372 for each in SearchType::ALL {
373 let counted = match self.count_of(each, &query, &reader).await {
374 Ok(count) => count,
375 Err(error) => {
376 worker::console_error!("search: counting {} failed: {error}", each.as_str());
377 0
378 }
379 };
380 results.counts.set(each, counted);
381 }
382 // One more than a page, to know whether there is another.
383 let window = Page { limit: per_page + 1, offset: (page - 1) * per_page };
384 let terms = query.highlights();
385 let found = match self.page_of(kind, &query, &reader, window, &terms).await {
386 Ok(found) => found,
387 Err(error) => {
388 // Most often a query the index could not read.
389 worker::console_error!("search: {} failed for {:?}: {error}", kind.as_str(), results.query);
390 results.notes.push("Part of that query could not be read. Try fewer symbols, or put them in quotes.".into());
391 Vec::new()
392 }
393 };
394 let before = found.len();
395 let mut hits = self.admit(&a.viewer, found).await;
396 // What the second check dropped is not counted either.
397 let dropped = (before - hits.len()) as u32;
398 results.counts.set(kind, results.counts.get(kind).saturating_sub(dropped));
399 results.more = hits.len() > per_page as usize;
400 hits.truncate(per_page as usize);
401 results.hits = hits;
402 Ok(Outcome::Ok(results))
403 }
404
405 /// A few repositories, issues, pull requests and people, for the
406 /// command palette. Never code: that is a full search away.
407 pub async fn suggest(&self, a: SuggestArgs) -> Result<Vec<Hit>> {
408 let query = Query::parse(&a.query);
409 if query.prose_match().is_none() {
410 return Ok(Vec::new());
411 }
412 let reader = Reader::of(&a.viewer);
413 let page = Page { limit: SUGGEST_EACH, offset: 0 };
414 let terms = query.highlights();
415 let mut found = Vec::new();
416 for kind in [SearchType::Repositories, SearchType::Issues, SearchType::Pulls, SearchType::People] {
417 match self.page_of(kind, &query, &reader, page, &terms).await {
418 Ok(more) => found.extend(more),
419 Err(error) => worker::console_error!("search: suggesting {} failed: {error}", kind.as_str()),
420 }
421 }
422 Ok(self.admit(&a.viewer, found).await)
423 }
424
425 pub async fn explore(&self, a: ExploreArgs) -> Result<Explore> {
426 let newest = a.sort.as_deref() == Some("new");
427 let language = a.language.as_deref().map(str::trim).filter(|l| !l.is_empty()).map(normalize_language);
428 let topic = a.topic.as_deref().map(|t| t.trim().to_lowercase()).filter(|t| !t.is_empty());
429 let page = a.page.unwrap_or(1).clamp(1, MAX_PAGE);
430 let window = Page { limit: EXPLORE_PAGE + 1, offset: (page - 1) * EXPLORE_PAGE };
431 let rows = store::all::<RepoRow>(&self.db, &sql::explore(newest, language.as_deref(), topic.as_deref(), window)).await?;
432 let indexed: Vec<Indexed> = rows
433 .iter()
434 .map(|row| Indexed { repo_id: row.repo_id.clone(), namespace: row.namespace.clone(), private: row.private != 0 })
435 .collect();
436 // Explore is for everyone: checked as if signed out, whoever asks.
437 let keep = self.readable(&None, &indexed).await;
438 let mut repos: Vec<ExploreRepo> = rows
439 .into_iter()
440 .filter(|row| keep.contains(&row.repo_id))
441 .map(|row| ExploreRepo {
442 namespace: row.namespace,
443 name: row.name,
444 description: row.description,
445 topics: split_topics(&row.topics),
446 language: row.language,
447 created_at: row.created_at,
448 pushed_at: row.pushed_at,
449 })
450 .collect();
451 let more = repos.len() > EXPLORE_PAGE as usize;
452 repos.truncate(EXPLORE_PAGE as usize);
453 let (languages, topics) = if page == 1 {
454 (
455 store::all::<Facet>(&self.db, &Sql { text: sql::LANGUAGE_FACETS.into(), params: vec![] }).await?,
456 store::all::<Facet>(&self.db, &Sql { text: sql::TOPIC_FACETS.into(), params: vec![] }).await?,
457 )
458 } else {
459 (Vec::new(), Vec::new())
460 };
461 Ok(Explore { repos, languages, topics, page, more })
462 }
463}
464
465#[cfg(test)]
466mod tests {
467 use super::*;
468
469 #[test]
470 fn paths_are_safe_in_addresses() {
471 assert_eq!(encode_path("src/main.rs"), "src/main.rs");
472 assert_eq!(encode_path("docs/a b#c?.md"), "docs/a%20b%23c%3F.md");
473 assert_eq!(encode_path("src/ünï.rs"), "src/%C3%BCn%C3%AF.rs");
474 }
475
476 #[test]
477 fn labels_and_topics_are_split() {
478 assert_eq!(split_labels("|bug|good first issue|"), vec!["bug", "good first issue"]);
479 assert!(split_labels("|").is_empty());
480 assert_eq!(split_topics("cli rust"), vec!["cli", "rust"]);
481 }
482}