Something went wrong. Try again.
Monorepo for Tangled tangled.org
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514use std::collections::{HashMap, HashSet};
use axum::extract::State;use bobbin_codesearch::{CodeSearchError, FileMatch, Range, extract_did};use futures::{StreamExt as _, stream};use jacquard_axum::service_auth::ExtractServiceAuth;use jacquard_axum::{ExtractXrpc, XrpcResponse};use jacquard_common::IntoStatic as _;use jacquard_common::types::did::Did;use lexicons::org_tangled::temp::search::search_code::{ self, Chunk, FileResult, Highlight, SearchCodeOutput,};
use crate::{AppState, FETCH_CONCURRENCY, XrpcError, view::build_repo_view_basic};
const DEFAULT_LIMIT: i64 = 50;const MAX_LIMIT: i64 = 100;
pub(crate) async fn search_code( State(state): State<AppState>, ExtractServiceAuth(auth): ExtractServiceAuth, ExtractXrpc(params): ExtractXrpc<search_code::SearchCodeRequest>,) -> Result<XrpcResponse<search_code::SearchCodeRequest>, XrpcError> { let viewer = auth.did().into_static(); search_code_inner(&state, Some(&viewer), params).await}
async fn search_code_inner( state: &AppState, viewer: Option<&Did>, params: search_code::SearchCode,) -> Result<XrpcResponse<search_code::SearchCodeRequest>, XrpcError> { let Some(codesearch) = state.codesearch.as_ref() else { return Err(XrpcError::NotImplemented("code search is not configured")); };
let q = params.q.trim(); if q.is_empty() { return Err(XrpcError::InvalidParams("q must not be empty".into())); } // scope by the repo's own did (meta.did) let query = match ¶ms.repo { Some(did) => format!("meta.did:{did} {q}"), None => q.to_owned(), };
let limit = params.limit.unwrap_or(DEFAULT_LIMIT).clamp(1, MAX_LIMIT) as usize; let offset = params .cursor .as_deref() .map(str::parse::<usize>) .transpose() .map_err(|e| XrpcError::InvalidParams(format!("cursor: {e}")))? .unwrap_or(0);
let _permit = state.heavy_permit()?; let page = codesearch .search(&query, offset, limit) .await .map_err(map_codesearch_err)?;
let mut matches: Vec<(Did, FileMatch)> = page .files .into_iter() .filter_map(|file| { let did = page .repo_urls .get(&file.repository) .and_then(|template| extract_did(template))?; Some((did, file)) }) .collect();
if let Some(wanted) = ¶ms.repo { matches.retain(|(did, _)| did == wanted); }
let views = hydrate_repos(state, viewer, &matches).await;
let mut results = Vec::with_capacity(matches.len()); for (did, file) in &matches { let Some(repo) = views.get(did).cloned() else { continue; }; let mut ranges = None; let mut chunks = Vec::new(); for chunk in &file.chunk_matches { if chunk.file_name { ranges = Some(chunk_highlights(&file.file_name, 1, &chunk.ranges)); break; } let line_start = chunk.content_start.line_number; chunks.push(Chunk { highlights: Some(chunk_highlights( &chunk.content, line_start.into(), &chunk.ranges, )), content: chunk.content.clone().into(), line_start: line_start.into(), extra_data: Default::default(), }); } results.push(FileResult { repo, path: file.file_name.clone().into(), commit: file.version.clone().into(), branches: (!file.branches.is_empty()) .then(|| file.branches.iter().map(|b| b.as_str().into()).collect()), language: (!file.language.is_empty()).then(|| file.language.as_str().into()), ranges, chunks: (!chunks.is_empty()).then_some(chunks), extra_data: Default::default(), }); }
Ok(XrpcResponse(SearchCodeOutput { cursor: page .has_more .then(|| (offset.saturating_add(limit)).to_string().into()), results, extra_data: Default::default(), }))}
async fn hydrate_repos( state: &AppState, viewer: Option<&Did>, matches: &[(Did, FileMatch)],) -> HashMap<Did, lexicons::sh_tangled::repo::RepoViewBasic> { let mut seen = HashSet::new(); let dids: Vec<Did> = matches .iter() .filter(|(did, _)| seen.insert(did.clone())) .map(|(did, _)| did.clone()) .collect();
stream::iter(dids.into_iter().map(|did| async move { match build_repo_view_basic(state, viewer, &did).await { Ok(view) => Some((did, view)), Err(err) => { tracing::warn!(%did, error = %err, "skipping code search result, repo hydration failed"); None } } })) .buffered(FETCH_CONCURRENCY) .filter_map(std::future::ready) .collect() .await}
/// Maps zoekt's 1-based (line, rune-column) ranges onto byte offsets within the chunk's content.fn chunk_highlights(content: &str, start_line: i64, ranges: &[Range]) -> Vec<Highlight> { let mut offset = 0usize; let lines: Vec<(usize, &str)> = content .split_inclusive('\n') .map(|line| { let start = offset; offset += line.len(); (start, line) }) .collect(); let start_line = start_line.max(1);
// maps a 1-based (line, rune column) to a byte offset in content let byte_at = |line: u32, column: u32| -> Option<usize> { let idx = usize::try_from(i64::from(line) - start_line).ok()?; let (offset, text) = *lines.get(idx)?; let col = column.saturating_sub(1) as usize; // a column past the end of the line clamps to just before its newline Some( offset + text .char_indices() .nth(col) .map_or_else(|| text.trim_end_matches('\n').len(), |(byte, _)| byte), ) };
ranges .iter() .filter_map(|range| { let start = byte_at(range.start.line_number, range.start.column)?; let end = byte_at(range.end.line_number, range.end.column)?; (start < end).then(|| Highlight { start: start as i64, end: end as i64, extra_data: Default::default(), }) }) .collect()}
fn map_codesearch_err(err: CodeSearchError) -> XrpcError { use CodeSearchError as E; match err { E::BadQuery(e) => XrpcError::InvalidParams(format!("query: {e}")), e @ (E::Network(_) | E::Upstream { .. }) => { XrpcError::UpstreamUnavailable(format!("code search: {e}")) } e @ E::Decode(_) => XrpcError::InvalidRecord(format!("code search: {e}")), e @ (E::BadScheme(_) | E::Build(_) | E::BadAuthorization) => { XrpcError::Internal(format!("code search: {e}")) } }}
#[cfg(test)]mod tests { use super::*; use bobbin_codesearch::{CodeSearch, Location}; use bobbin_edge_index::{CoverageWatch, EdgeStore}; use bobbin_record_lru::{CacheCapacity, LruRecordStore}; use bobbin_resolver::RepoIdResolver; use bobbin_runtime::{RuntimeHasher, SystemClock}; use bobbin_search::{DEFAULT_WRITER_HEAP_BYTES, SearchIndex, SearchReader}; use bobbin_slingshot_client::SlingshotClient; use jacquard_common::DefaultStr; use jacquard_common::types::string::AtUri; use serde_json::{Value, json}; use std::sync::Arc; use url::Url; use wiremock::matchers::{method, path, query_param}; use wiremock::{Mock, MockServer, ResponseTemplate};
const CID: &str = "bafyreieqygohnz2zqyvtvktbjpvhutphobcmbsnt4q5lc36ri7vpcmoz4i"; const OWNER: &str = "did:plc:scallop"; const REPO: &str = "did:plc:limpet";
fn did(s: &str) -> Did { Did::new_owned(s).unwrap() }
fn range(sl: u32, sc: u32, el: u32, ec: u32) -> Range { Range { start: Location { line_number: sl, column: sc, }, end: Location { line_number: el, column: ec, }, } }
fn params(q: &str) -> search_code::SearchCode { search_code::SearchCode { q: q.into(), repo: None, limit: None, cursor: None, } }
/// A zoekt `FileMatch` with one content chunk. `Content` is base64 on the wire, since the /// Go field is a `[]byte`. fn content_match(name: &str, line: u32, content: &str, cols: (u32, u32)) -> Value { use base64::Engine as _; json!({ "FileName": name, "Repository": "repo-7", "Version": "abc123", "Language": "Rust", "Branches": ["main"], "ChunkMatches": [{ "Content": base64::engine::general_purpose::STANDARD.encode(content), "ContentStart": {"LineNumber": line, "Column": 1}, "Ranges": [{ "Start": {"LineNumber": line, "Column": cols.0}, "End": {"LineNumber": line, "Column": cols.1}, }], }], }) }
async fn mock_zoekt(server: &MockServer, files: Value) { Mock::given(method("POST")) .and(path("/api/search")) .respond_with(ResponseTemplate::new(200).set_body_json(json!({ "Result": { "Files": files, "RepoURLs": { "repo-7": format!("https://tangled.org/{REPO}/blob/{{{{.Version}}}}/{{{{.Path}}}}"), }, } }))) .mount(server) .await; }
/// Slingshot and zoekt both point at `server`. When `hydratable`, the repo record is /// mounted so `build_repo_view_basic` resolves. async fn app(server: &MockServer, hydratable: bool) -> AppState { let base = Url::parse(&server.uri()).unwrap(); if hydratable { Mock::given(method("GET")) .and(path("/xrpc/com.atproto.repo.getRecord")) .and(query_param("repo", REPO)) .and(query_param("collection", "org.tangled.repo.manifest")) .and(query_param("rkey", "self")) .respond_with(ResponseTemplate::new(200).set_body_json(json!({ "uri": format!("at://{REPO}/org.tangled.repo.manifest/self"), "cid": CID, "value": { "$type": "org.tangled.repo.manifest", "declaration": { "owner": OWNER, "name": "core" }, "createdAt": "2026-05-01T00:00:00Z", }, }))) .mount(server) .await; } let state = AppState::new( Arc::new(LruRecordStore::new(CacheCapacity::from_bytes(64 * 1024))), SlingshotClient::with_default_http(base.clone()).unwrap(), Arc::new(EdgeStore::new(RuntimeHasher::default())), Arc::new(CoverageWatch::new()), Arc::new( SearchIndex::new(DEFAULT_WRITER_HEAP_BYTES, Arc::new(SystemClock::new())).unwrap(), ) as Arc<dyn SearchReader>, Arc::new(RepoIdResolver::detached(RuntimeHasher::default())), Arc::new(crate::default_directory()), ); if hydratable { let uri = |s: String| AtUri::<DefaultStr>::new_owned(s).unwrap(); let manifest = uri(format!("at://{REPO}/org.tangled.repo.manifest/self")); state .claims .upsert_manifest(&manifest, OWNER, "core", Vec::new()); let declaration = uri(format!("at://{OWNER}/org.tangled.repo.declaration/core")); state.claims.upsert_declaration(&declaration, REPO); } state.with_codesearch(Some(Arc::new(CodeSearch::new(&base).unwrap()))) }
async fn run(state: &AppState, p: search_code::SearchCode) -> Result<Value, XrpcError> { search_code_inner(state, None, p) .await .map(|resp| serde_json::to_value(resp.0).unwrap()) }
#[tokio::test] async fn answers_not_implemented_when_zoekt_is_unconfigured() { let server = MockServer::start().await; let state = app(&server, true).await.with_codesearch(None); let err = run(&state, params("fn main")).await.expect_err("must 501"); assert!(matches!(err, XrpcError::NotImplemented(_)), "got {err}"); }
#[tokio::test] async fn returns_hydrated_file_results() { let server = MockServer::start().await; // "é" is two bytes, so the rune columns zoekt reports are not byte offsets mock_zoekt( &server, json!([content_match("src/main.rs", 12, "// héllo\n", (4, 9))]), ) .await; let state = app(&server, true).await;
let body = run(&state, params("fn main")).await.expect("ok"); let result = &body["results"][0]; assert_eq!(result["path"], "src/main.rs"); assert_eq!(result["commit"], "abc123"); assert_eq!(result["language"], "Rust"); assert_eq!(result["branches"][0], "main"); // the repo view came out of the mounted record, keyed by the DID scraped from RepoURLs assert_eq!(result["repo"]["did"], REPO); assert_eq!(result["repo"]["slug"], "core"); assert_eq!(result["repo"]["owner"]["did"], OWNER); // base64 decoded, and rune columns 4..9 landed on bytes 3..9 let chunk = &result["chunks"][0]; assert_eq!(chunk["content"], "// héllo\n"); assert_eq!(chunk["lineStart"], 12); assert_eq!(chunk["highlights"][0]["start"], 3); assert_eq!(chunk["highlights"][0]["end"], 9); // one page, so no cursor assert!(body.get("cursor").is_none(), "got {body}"); }
#[tokio::test] async fn scopes_by_repo_did_and_pages_with_an_offset_cursor() { let server = MockServer::start().await; // three docs for a limit of one: zoekt is asked for offset+limit+1, and the extra doc // is what tells us a further page exists mock_zoekt( &server, json!([ content_match("a.rs", 1, "one\n", (1, 4)), content_match("b.rs", 1, "two\n", (1, 4)), content_match("c.rs", 1, "three\n", (1, 4)), ]), ) .await; let state = app(&server, true).await;
let body = run( &state, search_code::SearchCode { q: "fn main".into(), repo: Some(did(REPO)), limit: Some(1), cursor: Some("1".into()), }, ) .await .expect("ok");
// windowed to the second doc assert_eq!(body["results"].as_array().unwrap().len(), 1); assert_eq!(body["results"][0]["path"], "b.rs"); assert_eq!(body["cursor"], "2");
let requests = server.received_requests().await.unwrap(); let sent: Value = serde_json::from_slice( &requests .iter() .find(|r| r.url.path() == "/api/search") .expect("zoekt was called") .body, ) .unwrap(); assert_eq!(sent["Q"], format!("meta.did:{REPO} fn main")); assert!(sent["Opts"]["ChunkMatches"].as_bool().unwrap()); assert_eq!(sent["Opts"]["MaxDocDisplayCount"], 3); // offset 1 + limit 1 + probe }
#[tokio::test] async fn drops_results_whose_repo_cannot_be_hydrated() { let server = MockServer::start().await; mock_zoekt(&server, json!([content_match("a.rs", 1, "one\n", (1, 4))])).await; // no repo record mounted and nothing observed, so hydration misses let state = app(&server, false).await; let body = run(&state, params("fn main")).await.expect("ok"); assert!(body["results"].as_array().unwrap().is_empty(), "got {body}"); }
#[tokio::test] async fn rejects_a_blank_query_and_a_non_numeric_cursor() { let server = MockServer::start().await; let state = app(&server, true).await;
let err = run(&state, params(" ")).await.expect_err("must reject"); assert!(matches!(err, XrpcError::InvalidParams(_)), "got {err}");
let mut p = params("fn"); p.cursor = Some("abc".into()); let err = run(&state, p).await.expect_err("must reject"); match err { XrpcError::InvalidParams(m) => assert!(m.contains("cursor"), "got {m}"), other => panic!("got {other}"), } }
#[tokio::test] async fn maps_a_zoekt_query_rejection_to_invalid_params() { let server = MockServer::start().await; Mock::given(method("POST")) .and(path("/api/search")) .respond_with(ResponseTemplate::new(400).set_body_string("parse error")) .mount(&server) .await; let state = app(&server, true).await; let err = run(&state, params("(unbalanced")) .await .expect_err("must reject"); assert!(matches!(err, XrpcError::InvalidParams(_)), "got {err}"); }
#[test] fn maps_rune_columns_to_byte_offsets() { // "é" is two bytes, so rune columns on line 2 must not be read as byte offsets let content = "// héllo\nlet x = 1;\n"; let highlights = chunk_highlights( content, 7, &[ // line 7, runes 4..9 => "héllo", bytes 3..9 (one extra byte for é) range(7, 4, 7, 9), // line 8, runes 5..6 => "x", and line 8 starts at byte 10 range(8, 5, 8, 6), ], ); let spans: Vec<&str> = highlights .iter() .map(|h| &content[h.start as usize..h.end as usize]) .collect(); assert_eq!(spans, vec!["héllo", "x"]); assert_eq!((highlights[0].start, highlights[0].end), (3, 9)); }
#[test] fn drops_ranges_outside_the_chunk() { let content = "let x = 1;\n"; // before the chunk, after the chunk, and empty let highlights = chunk_highlights( content, 7, &[range(6, 1, 6, 4), range(9, 1, 9, 4), range(7, 3, 7, 3)], ); assert!(highlights.is_empty(), "got {highlights:?}"); }
#[test] fn clamps_a_column_past_the_end_of_the_line() { let content = "ab\n"; // rune column 99 has no byte; clamp to the offset before the newline let highlights = chunk_highlights(content, 1, &[range(1, 1, 1, 99)]); assert_eq!((highlights[0].start, highlights[0].end), (0, 2)); }}