194c44c
use std::collections::{HashMap, HashSet, VecDeque};
use std::path::Path;
use std::sync::Arc;
use std::time::{Duration, Instant};
use config::search::SearchConfig;
use export::AppDatabase;
use models::chat::messages::{self, ChatMessageType, FLAG_EPHEMERAL};
use models::dir::dirs;
use models::issue::{comment, issue_item, issue_label, issue_relation_label, issues};
use models::pull_request::{
pull_request_comment, pull_request_label, pull_request_review, pull_requests,
};
use models::repo::{repo_collaborations, repos};
use models::room::{room_bans, room_members, rooms};
use models::search::repo_state;
use models::space::{space_members, spaces};
use models::users::user;
use reqwest::{Client, StatusCode};
use sea_orm::sea_query::OnConflict;
use sea_orm::{
ColumnTrait, ConnectionTrait, DatabaseBackend, EntityTrait, QueryFilter, QueryOrder,
QuerySelect, Set, Statement, TransactionTrait,
};
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use tokio::sync::Semaphore;
use tokio::task::{JoinHandle, JoinSet};
use tracing::{debug, warn};
use utoipa::{IntoParams, ToSchema};
use uuid::Uuid;
use crate::indexer::{
GitCatFile, git_output, parse_ls_tree, parse_name_status, rev_exists, text_chunks,
};
use crate::parser::{language_label, symbols_for};
#[derive(Debug, thiserror::Error)]
pub enum SearchError {
#[error("search is disabled")]
Disabled,
#[error("invalid search query: {0}")]
Invalid(String),
#[error("search index task failed [{code}]: {message}")]
Task { code: String, message: String },
#[error("search service request failed: {0}")]
Http(#[from] reqwest::Error),
#[error("search service returned {status}: {body}")]
Remote { status: StatusCode, body: String },
#[error("database error: {0}")]
Db(#[from] sea_orm::DbErr),
#[error("git indexing command failed: {0}")]
Git(String),
#[error("search serialization failed: {0}")]
Json(#[from] serde_json::Error),
#[error("search I/O failed: {0}")]
Io(#[from] std::io::Error),
#[error("search task timed out")]
TimedOut,
}
fn is_duplicate_index(error: &SearchError) -> bool {
matches!(error, SearchError::Task { code, .. } if code == "index_already_exists")
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, ToSchema)]
#[serde(rename_all = "snake_case")]
pub enum SearchKind {
Code,
Symbol,
Issue,
Pull,
Message,
}
impl SearchKind {
fn parse_list(value: Option<&str>) -> Result<Vec<Self>, SearchError> {
let Some(value) = value.filter(|v| !v.trim().is_empty()) else {
return Ok(vec![
Self::Code,
Self::Symbol,
Self::Issue,
Self::Pull,
Self::Message,
]);
};
let mut kinds = Vec::new();
for value in value.split(',').map(str::trim).filter(|v| !v.is_empty()) {
let kind = match value {
"code" => Self::Code,
"symbol" | "symbols" => Self::Symbol,
"issue" | "issues" => Self::Issue,
"pull" | "pulls" | "pr" => Self::Pull,
"message" | "messages" => Self::Message,
other => return Err(SearchError::Invalid(format!("unknown kind `{other}`"))),
};
if !kinds.contains(&kind) {
kinds.push(kind);
}
}
if kinds.is_empty() {
return Err(SearchError::Invalid("kinds must not be empty".to_string()));
}
Ok(kinds)
}
fn index_suffix(self) -> &'static str {
match self {
Self::Code => "code",
Self::Symbol => "symbols",
Self::Issue => "issues",
Self::Pull => "pulls",
Self::Message => "messages",
}
}
fn as_str(self) -> &'static str {
match self {
Self::Code => "code",
Self::Symbol => "symbol",
Self::Issue => "issue",
Self::Pull => "pull",
Self::Message => "message",
}
}
}
#[derive(Debug, Clone, Deserialize, IntoParams)]
pub struct SearchQuery {
pub q: String,
pub kinds: Option<String>,
pub space_name: Option<String>,
pub repo_name: Option<String>,
/// 限定单个看板（工单类命中）与单个频道（消息类命中）。
///
/// 全局搜索从仓库页或频道页打开时，界面会把当前所在位置作为默认约束传进来，
/// 「在这里搜」和「到处搜」是两种不同意图；把维度做成查询参数，前端就不必在
/// 结果里再筛一遍。
pub board_id: Option<Uuid>,
pub room_id: Option<Uuid>,
pub language: Option<String>,
pub state: Option<String>,
pub limit: Option<usize>,
pub offset: Option<usize>,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct SearchHit {
pub kind: SearchKind,
pub id: String,
pub title: String,
pub snippet: String,
pub space_name: Option<String>,
pub repo_name: Option<String>,
pub board_id: Option<Uuid>,
pub number: Option<i32>,
pub revision: Option<String>,
pub path: Option<String>,
pub line_start: Option<u32>,
pub line_end: Option<u32>,
pub language: Option<String>,
pub state: Option<String>,
pub symbol_id: Option<String>,
pub symbol_name: Option<String>,
pub symbol_kind: Option<String>,
pub definition: Option<bool>,
pub room_id: Option<Uuid>,
pub room_name: Option<String>,
pub thread_id: Option<Uuid>,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct SearchPage {
pub items: Vec<SearchHit>,
pub total: usize,
pub limit: usize,
pub offset: usize,
}
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
struct SearchDocument {
id: String,
kind: String,
repo_id: Option<String>,
space_id: Option<String>,
space_name: Option<String>,
repo_name: Option<String>,
board_id: Option<String>,
number: Option<i32>,
revision: Option<String>,
path: Option<String>,
line_start: Option<u32>,
line_end: Option<u32>,
language: Option<String>,
state: Option<String>,
symbol_id: Option<String>,
symbol_name: Option<String>,
symbol_kind: Option<String>,
definition: Option<bool>,
room_id: Option<String>,
room_name: Option<String>,
thread_id: Option<String>,
title: String,
content: String,
#[serde(default, skip_serializing, rename = "_matchesPosition")]
matches: HashMap<String, Vec<MatchPosition>>,
}
#[derive(Debug, Clone, Deserialize)]
struct MatchPosition {
#[serde(default)]
start: usize,
#[serde(default)]
length: usize,
}
fn char_floor(text: &str, index: usize) -> usize {
let mut value = index.min(text.len());
while value > 0 && !text.is_char_boundary(value) {
value -= 1;
}
value
}
fn char_ceil(text: &str, index: usize) -> usize {
let mut value = index.min(text.len());
while value < text.len() && !text.is_char_boundary(value) {
value += 1;
}
value
}
// 片段从命中的那一行开始，前端只渲染一行，命中必须出现在开头。
fn window_snippet(text: &str, start: usize, length: usize) -> String {
if start >= text.len() {
return compact_snippet(text);
}
let hit_start = char_floor(text, start);
let hit_end = char_ceil(text, (start + length).min(text.len()));
let from = text[..hit_start].rfind('\n').map_or(0, |offset| offset + 1);
let mut until = char_ceil(text, (hit_end + 300).min(text.len()));
if let Some(offset) = text[until..].find('\n') {
until = char_ceil(text, until + offset);
}
let normalized = text[from..until]
.split_whitespace()
.collect::<Vec<_>>()
.join(" ");
let prefix = if from > 0 { "…" } else { "" };
let suffix = if until < text.len() { "…" } else { "" };
format!("{prefix}{normalized}{suffix}")
}
fn symbol_label(kind: &str) -> &'static str {
match kind {
"function_item" | "function_declaration" | "function_definition" => "function",
"method_definition" | "method_declaration" => "method",
"struct_item" | "struct_specifier" => "struct",
"enum_item" | "enum_specifier" => "enum",
"trait_item" => "trait",
"impl_item" => "impl",
"type_item" | "type_declaration" => "type alias",
"class_definition" | "class_declaration" => "class",
"interface_declaration" => "interface",
"module" | "namespace_definition" => "module",
"reference" => "reference",
_ => "symbol",
}
}
fn matched_snippet(document: &SearchDocument) -> String {
let matched = document
.matches
.get("content")
.and_then(|hits| hits.iter().min_by_key(|hit| hit.start));
match matched {
Some(matched) => window_snippet(&document.content, matched.start, matched.length),
None => compact_snippet(&document.content),
}
}
impl SearchDocument {
fn hit(self) -> SearchHit {
let kind = match self.kind.as_str() {
"message" => SearchKind::Message,
"symbol" => SearchKind::Symbol,
"issue" => SearchKind::Issue,
"pull" => SearchKind::Pull,
_ => SearchKind::Code,
};
let snippet = match (&kind, self.symbol_kind.as_deref()) {
(SearchKind::Symbol, Some(symbol_kind)) => symbol_label(symbol_kind).to_string(),
_ => matched_snippet(&self),
};
SearchHit {
kind,
id: self.id,
title: self.title,
snippet,
space_name: self.space_name,
repo_name: self.repo_name,
board_id: self.board_id.and_then(|id| Uuid::parse_str(&id).ok()),
number: self.number,
revision: self.revision,
path: self.path,
line_start: self.line_start,
line_end: self.line_end,
language: self.language,
state: self.state,
symbol_id: self.symbol_id,
symbol_name: self.symbol_name,
symbol_kind: self.symbol_kind,
definition: self.definition,
room_id: self.room_id.and_then(|id| Uuid::parse_str(&id).ok()),
room_name: self.room_name,
thread_id: self.thread_id.and_then(|id| Uuid::parse_str(&id).ok()),
}
}
}
#[derive(Clone)]
struct MeiliClient {
client: Client,
base: String,
api_key: String,
timeout: Duration,
configured: std::sync::Arc<tokio::sync::Mutex<HashSet<String>>>,
}
impl MeiliClient {
fn new(config: &SearchConfig) -> Result<Self, SearchError> {
let client = Client::builder()
.timeout(Duration::from_secs(config.task_timeout_secs))
.build()?;
Ok(Self {
client,
base: config.url.trim_end_matches('/').to_string(),
api_key: config.api_key.clone(),
timeout: Duration::from_secs(config.task_timeout_secs),
configured: std::sync::Arc::new(tokio::sync::Mutex::new(HashSet::new())),
})
}
fn request(&self, method: reqwest::Method, path: &str) -> reqwest::RequestBuilder {
self.client
.request(method, format!("{}{}", self.base, path))
.bearer_auth(&self.api_key)
}
async fn json(&self, request: reqwest::RequestBuilder) -> Result<Value, SearchError> {
let response = request.send().await?;
let status = response.status();
let body = response.text().await?;
if !status.is_success() {
return Err(SearchError::Remote { status, body });
}
Ok(serde_json::from_str(&body)?)
}
async fn index_exists(&self, index: &str) -> Result<bool, SearchError> {
let response = self
.request(reqwest::Method::GET, &format!("/indexes/{index}"))
.send()
.await?;
match response.status() {
StatusCode::OK => Ok(true),
StatusCode::NOT_FOUND => Ok(false),
status => {
let body = response.text().await?;
Err(SearchError::Remote { status, body })
}
}
}
async fn ensure_index(&self, index: &str, searchable: &[&str]) -> Result<(), SearchError> {
let mut configured = self.configured.lock().await;
if configured.contains(index) {
return Ok(());
}
if !self.index_exists(index).await? {
let response = self
.request(reqwest::Method::POST, "/indexes")
.json(&json!({ "uid": index, "primaryKey": "id" }))
.send()
.await?;
let status = response.status();
if status.is_success() {
let task: Value = response.json().await?;
if let Err(error) = self.wait_task(&task).await
&& !is_duplicate_index(&error)
{
return Err(error);
}
} else if status != StatusCode::CONFLICT {
let body = response.text().await?;
return Err(SearchError::Remote { status, body });
}
}
let task = self
.json(
self.request(reqwest::Method::PATCH, &format!("/indexes/{index}/settings"))
.json(&json!({
"searchableAttributes": searchable,
"filterableAttributes": ["repo_id", "space_id", "space_name", "repo_name", "board_id", "language", "state", "definition", "room_id", "path"],
"displayedAttributes": ["*"],
})),
)
.await?;
self.wait_task(&task).await?;
configured.insert(index.to_string());
Ok(())
}
async fn add_documents(
&self,
index: &str,
documents: &[SearchDocument],
) -> Result<(), SearchError> {
if documents.is_empty() {
return Ok(());
}
let task = self
.json(
self.request(
reqwest::Method::POST,
&format!("/indexes/{index}/documents"),
)
.json(documents),
)
.await?;
self.wait_task(&task).await
}
async fn delete_filter(&self, index: &str, filter: &str) -> Result<(), SearchError> {
let task = self
.json(
self.request(
reqwest::Method::POST,
&format!("/indexes/{index}/documents/delete"),
)
.json(&json!({ "filter": filter })),
)
.await?;
self.wait_task(&task).await
}
async fn delete_document(&self, index: &str, id: &str) -> Result<(), SearchError> {
let task = self
.json(self.request(
reqwest::Method::DELETE,
&format!("/indexes/{index}/documents/{id}"),
))
.await?;
self.wait_task(&task).await
}
async fn search(
&self,
index: &str,
query: &str,
filter: Option<String>,
limit: usize,
offset: usize,
) -> Result<(Vec<SearchDocument>, usize), SearchError> {
let value = self
.json(
self.request(reqwest::Method::POST, &format!("/indexes/{index}/search"))
.json(&json!({
"q": query,
"filter": filter,
"limit": limit,
"offset": offset,
"attributesToRetrieve": ["*"],
"showMatchesPosition": true,
})),
)
.await?;
let hits = serde_json::from_value(value.get("hits").cloned().unwrap_or_else(|| json!([])))?;
let total = value
.get("estimatedTotalHits")
.and_then(Value::as_u64)
.unwrap_or_default() as usize;
Ok((hits, total))
}
async fn wait_task(&self, task: &Value) -> Result<(), SearchError> {
let Some(id) = task
.get("taskUid")
.or_else(|| task.get("uid"))
.and_then(Value::as_u64)
else {
return Ok(());
};
self.wait_uid(id).await
}
async fn wait_uid(&self, id: u64) -> Result<(), SearchError> {
let deadline = Instant::now() + self.timeout;
loop {
let state = self
.json(self.request(reqwest::Method::GET, &format!("/tasks/{id}")))
.await?;
match state.get("status").and_then(Value::as_str) {
Some("succeeded") => return Ok(()),
Some("failed") | Some("canceled") => {
let error = state.get("error").and_then(Value::as_object);
let code = error
.and_then(|error| error.get("code"))
.and_then(Value::as_str)
.unwrap_or("task_failed")
.to_string();
let message = error
.and_then(|error| error.get("message"))
.and_then(Value::as_str)
.unwrap_or("Meilisearch task failed")
.to_string();
return Err(SearchError::Task { code, message });
}
_ if Instant::now() >= deadline => return Err(SearchError::TimedOut),
_ => tokio::time::sleep(Duration::from_millis(100)).await,
}
}
}
async fn submit_json(
&self,
method: reqwest::Method,
path: &str,
body: &Value,
) -> Result<Option<u64>, SearchError> {
let value = self.json(self.request(method, path).json(body)).await?;
Ok(value
.get("taskUid")
.or_else(|| value.get("uid"))
.and_then(Value::as_u64))
}
async fn submit_documents(
&self,
index: &str,
documents: &[SearchDocument],
) -> Result<Option<u64>, SearchError> {
if documents.is_empty() {
return Ok(None);
}
self.submit_json(
reqwest::Method::POST,
&format!("/indexes/{index}/documents"),
&serde_json::to_value(documents)?,
)
.await
}
async fn submit_delete_filter(
&self,
index: &str,
filter: &str,
) -> Result<Option<u64>, SearchError> {
self.submit_json(
reqwest::Method::POST,
&format!("/indexes/{index}/documents/delete"),
&json!({ "filter": filter }),
)
.await
}
}
#[derive(Clone)]
pub struct SearchService {
db: std::sync::Arc<AppDatabase>,
config: SearchConfig,
meili: Option<MeiliClient>,
}
#[derive(Clone)]
struct RepoContext {
repo: repos::Model,
space: spaces::Model,
}
impl SearchService {
pub fn from_parts(
db: std::sync::Arc<AppDatabase>,
config: SearchConfig,
) -> Result<Self, SearchError> {
let meili = config
.enabled
.then(|| MeiliClient::new(&config))
.transpose()?;
Ok(Self { db, config, meili })
}
pub fn enabled(&self) -> bool {
self.meili.is_some()
}
pub fn config(&self) -> &SearchConfig {
&self.config
}
pub fn database(&self) -> &std::sync::Arc<AppDatabase> {
&self.db
}
fn meili(&self) -> Result<&MeiliClient, SearchError> {
self.meili.as_ref().ok_or(SearchError::Disabled)
}
fn index(&self, kind: SearchKind) -> String {
format!("{}_{}", self.config.index_prefix, kind.index_suffix())
}
async fn ensure_indexes(&self) -> Result<(), SearchError> {
let meili = self.meili()?;
meili
.ensure_index(
&self.index(SearchKind::Code),
&["path", "content", "language"],
)
.await?;
meili
.ensure_index(
&self.index(SearchKind::Symbol),
&["symbol_name", "path", "content"],
)
.await?;
meili
.ensure_index(&self.index(SearchKind::Issue), &["title", "content"])
.await?;
meili
.ensure_index(&self.index(SearchKind::Pull), &["title", "content"])
.await?;
meili
.ensure_index(&self.index(SearchKind::Message), &["content"])
.await
}
pub async fn search(
&self,
caller: Option<i64>,
query: SearchQuery,
) -> Result<SearchPage, SearchError> {
self.search_scoped(caller, query, true).await
}
// definitions_only 决定符号检索是否只返回定义：符号索引同时登记了标识符引用，
// 不加这层过滤时引用数量会把定义和代码命中完全淹没。
async fn search_scoped(
&self,
caller: Option<i64>,
query: SearchQuery,
definitions_only: bool,
) -> Result<SearchPage, SearchError> {
let q = query.q.trim();
if q.is_empty() {
return Err(SearchError::Invalid("q must not be empty".to_string()));
}
let limit = query.limit.unwrap_or(20).clamp(1, 50);
let offset = query.offset.unwrap_or(0);
let kinds = SearchKind::parse_list(query.kinds.as_deref())?;
let visible_repos = self.visible_repos(caller).await?;
let visible_repo_ids: Vec<String> = visible_repos
.iter()
.map(|ctx| ctx.repo.id.to_string())
.collect();
let visible_spaces = self.visible_space_ids(caller).await?;
// 只有真的要查消息时才去算房间可见性，代码/工单检索不必付这份开销。
let visible_rooms = if kinds.contains(&SearchKind::Message) {
self.visible_room_ids(caller).await?
} else {
Vec::new()
};
let meili = self.meili()?;
let mut buckets: Vec<Vec<SearchHit>> = Vec::new();
let mut total = 0usize;
for kind in kinds {
if matches!(kind, SearchKind::Code | SearchKind::Symbol) && visible_repo_ids.is_empty()
{
continue;
}
if kind == SearchKind::Pull && visible_repo_ids.is_empty() {
continue;
}
if kind == SearchKind::Issue && visible_spaces.is_empty() {
continue;
}
if kind == SearchKind::Message && visible_rooms.is_empty() {
continue;
}
let filter = build_filter(
kind,
&FilterOptions {
definitions_only,
repo_ids: &visible_repo_ids,
space_ids: &visible_spaces,
room_ids: &visible_rooms,
space_name: query.space_name.as_deref(),
repo_name: query.repo_name.as_deref(),
board_id: query.board_id.map(|value| value.to_string()),
room_id: query.room_id.map(|value| value.to_string()),
language: query.language.as_deref(),
state: query.state.as_deref(),
},
);
let (hits, count) = meili
.search(&self.index(kind), q, filter, limit, offset)
.await?;
total += count;
buckets.push(hits.into_iter().map(SearchDocument::hit).collect());
}
Ok(SearchPage {
items: merge_buckets(buckets, limit),
total,
limit,
offset,
})
}
pub async fn symbols(
&self,
caller: Option<i64>,
space_name: String,
repo_name: String,
q: String,
limit: Option<usize>,
offset: Option<usize>,
) -> Result<SearchPage, SearchError> {
self.search(
caller,
SearchQuery {
q,
kinds: Some("symbol".to_string()),
space_name: Some(space_name),
repo_name: Some(repo_name),
board_id: None,
room_id: None,
language: None,
state: None,
limit,
offset,
},
)
.await
}
pub async fn references(
&self,
caller: Option<i64>,
space_name: String,
repo_name: String,
symbol_name: String,
limit: Option<usize>,
offset: Option<usize>,
) -> Result<SearchPage, SearchError> {
self.search_scoped(
caller,
SearchQuery {
q: symbol_name,
kinds: Some("symbol".to_string()),
space_name: Some(space_name),
repo_name: Some(repo_name),
board_id: None,
room_id: None,
language: None,
state: None,
limit,
offset,
},
false,
)
.await
}
pub async fn index_repo(&self, repo_id: Uuid, repo_root: &Path) -> Result<(), SearchError> {
self.ensure_indexes().await?;
let Some(context) = self.repo_context(repo_id).await? else {
warn!(%repo_id, "search code index skipped: repository or space not found");
return Ok(());
};
let repo_dir = repo_root.join(repo_id.to_string());
let sha = match git_output(
&repo_dir,
&[
"rev-parse",
&format!("refs/heads/{}", context.repo.default_branch),
],
)
.await
{
Ok(value) => String::from_utf8_lossy(&value).trim().to_string(),
Err(error) => {
warn!(%repo_id, error = %error, "search code index skipped: default branch not found");
return Ok(());
}
};
if sha.is_empty() {
warn!(%repo_id, "search code index skipped: default branch is empty");
return Ok(());
}
let started = Instant::now();
let meili = self.meili()?.clone();
let code_index = self.index(SearchKind::Code);
let symbol_index = self.index(SearchKind::Symbol);
let txn = self.db.write().begin().await?;
txn.query_one_raw(Statement::from_string(
DatabaseBackend::Postgres,
format!("SELECT pg_advisory_xact_lock(hashtext('search:repo:{repo_id}'))"),
))
.await?;
let previous = repo_state::Entity::find_by_id(repo_id)
.one(&txn)
.await?
.map(|row| row.revision);
let mut full = !self.config.incremental;
let mut changed: Vec<String> = Vec::new();
let mut deleted: Vec<String> = Vec::new();
if !full {
match previous.as_deref() {
None => full = true,
Some(old) if old == sha => {
txn.commit().await?;
debug!(%repo_id, "search repo index already current");
return Ok(());
}
Some(old) => {
if rev_exists(&repo_dir, old).await {
match git_output(
&repo_dir,
&["diff", "--name-status", "-z", "--no-renames", old, &sha],
)
.await
{
Ok(bytes) => {
let (paths, removed) = parse_name_status(&bytes);
if paths.len() + removed.len()
> self.config.incremental_max_changed_files
{
full = true;
} else {
changed = paths;
deleted = removed;
}
}
Err(error) => {
warn!(%repo_id, error = %error, "search repo diff failed; falling back to full rebuild");
full = true;
}
}
} else {
full = true;
}
}
}
}
let tree = git_output(&repo_dir, &["ls-tree", "-r", "-z", "--long", &sha]).await?;
let mut entries = parse_ls_tree(&tree);
if !full {
let wanted: HashSet<&str> = changed.iter().map(String::as_str).collect();
entries.retain(|entry| wanted.contains(entry.path.as_str()));
}
let mut task_uids: Vec<u64> = Vec::new();
if full {
if let Some(uid) = meili
.submit_delete_filter(&code_index, &format!("repo_id = \"{repo_id}\""))
.await?
{
task_uids.push(uid);
}
if let Some(uid) = meili
.submit_delete_filter(&symbol_index, &format!("repo_id = \"{repo_id}\""))
.await?
{
task_uids.push(uid);
}
} else {
let mut paths: Vec<String> = deleted.clone();
paths.extend(changed.iter().cloned());
for chunk in paths.chunks(DELETE_PATH_CHUNK) {
let filter = path_filter(repo_id, chunk);
if let Some(uid) = meili.submit_delete_filter(&code_index, &filter).await? {
task_uids.push(uid);
}
if let Some(uid) = meili.submit_delete_filter(&symbol_index, &filter).await? {
task_uids.push(uid);
}
}
}
let doc_context = Arc::new(DocContext {
repo_id,
space_id: context.space.id,
space_name: context.space.name.clone(),
repo_name: context.repo.name.clone(),
sha: sha.clone(),
chunk_lines: self.config.chunk_lines,
});
let max_file_bytes = self.config.max_file_bytes as u64;
let parse_threads = self.config.parse_threads.max(1);
let inflight_limit = parse_threads * 2;
let permits = Arc::new(Semaphore::new(parse_threads));
let mut handles: VecDeque<JoinHandle<(Vec<SearchDocument>, Vec<SearchDocument>)>> =
VecDeque::new();
let mut sink = IndexSink::new(
&meili,
&code_index,
&symbol_index,
self.config.batch_size.max(1),
);
let mut cat_file = if self.config.cat_file_batch {
Some(GitCatFile::spawn(&repo_dir, self.config.git_timeout_secs)?)
} else {
None
};
for entry in entries {
while handles.len() >= inflight_limit {
let Some(handle) = handles.pop_front() else {
break;
};
let (code_docs, symbol_docs) = handle.await.map_err(join_error)?;
sink.absorb(code_docs, symbol_docs).await?;
}
let Some(language) = language_label(&entry.path) else {
continue;
};
if entry.size > max_file_bytes {
continue;
}
let content = match cat_file.as_mut() {
Some(reader) => reader.read_blob(&entry.oid).await?,
None => git_output(&repo_dir, &["show", &format!("{sha}:{}", entry.path)])
.await
.map(Some)?,
};
let Some(content) = content else {
continue;
};
if content.contains(&0) {
continue;
}
let Ok(text) = String::from_utf8(content) else {
continue;
};
let Ok(permit) = permits.clone().acquire_owned().await else {
return Err(SearchError::Git("parse pool closed".to_string()));
};
let path = entry.path;
let parse_context = doc_context.clone();
handles.push_back(tokio::task::spawn_blocking(move || {
let _permit = permit;
parse_file(&parse_context, &path, language, &text)
}));
}
while let Some(handle) = handles.pop_front() {
let (code_docs, symbol_docs) = handle.await.map_err(join_error)?;
sink.absorb(code_docs, symbol_docs).await?;
}
sink.finish().await?;
task_uids.extend(std::mem::take(&mut sink.uids));
let wait_permits = Arc::new(Semaphore::new(self.config.meili_inflight.max(1)));
let mut waiters = JoinSet::new();
for uid in task_uids {
let Ok(permit) = wait_permits.clone().acquire_owned().await else {
return Err(SearchError::Git("search wait pool closed".to_string()));
};
let client = meili.clone();
waiters.spawn(async move {
let _permit = permit;
client.wait_uid(uid).await
});
}
while let Some(result) = waiters.join_next().await {
result.map_err(join_error)??;
}
repo_state::Entity::insert(repo_state::ActiveModel {
repo_id: Set(repo_id),
revision: Set(sha.clone()),
updated_at: Set(time::OffsetDateTime::now_utc()),
})
.on_conflict(
OnConflict::column(repo_state::Column::RepoId)
.update_columns([repo_state::Column::Revision, repo_state::Column::UpdatedAt])
.to_owned(),
)
.exec(&txn)
.await?;
txn.commit().await?;
if sink.capped() {
warn!(
%repo_id,
limit = MAX_CODE_DOCUMENTS,
"search code index reached document cap"
);
}
debug!(
%repo_id,
full,
files = sink.files(),
changed = changed.len(),
deleted = deleted.len(),
code_documents = sink.code_count(),
symbol_documents = sink.symbol_count(),
elapsed_ms = started.elapsed().as_millis() as u64,
"search repo index finished"
);
Ok(())
}
pub async fn index_issue(&self, issue_id: Uuid) -> Result<(), SearchError> {
self.ensure_indexes().await?;
let meili = self.meili()?;
let Some(issue) = issue_item::Entity::find_by_id(issue_id)
.one(self.db.read())
.await?
else {
return meili
.delete_document(&self.index(SearchKind::Issue), &issue_id.to_string())
.await;
};
let Some(board) = issues::Entity::find_by_id(issue.collection_id)
.one(self.db.read())
.await?
else {
return Ok(());
};
let Some(space) = self.space_for_dir(board.dir_id).await? else {
return Ok(());
};
let comments = comment::Entity::find()
.filter(comment::Column::IssueId.eq(issue_id))
.order_by_asc(comment::Column::CreatedAt)
.all(self.db.read())
.await?;
let label_ids: Vec<Uuid> = issue_relation_label::Entity::find()
.filter(issue_relation_label::Column::IssueId.eq(issue_id))
.all(self.db.read())
.await?
.into_iter()
.map(|row| row.label_id)
.collect();
let labels = if label_ids.is_empty() {
Vec::new()
} else {
issue_label::Entity::find()
.filter(issue_label::Column::Id.is_in(label_ids))
.all(self.db.read())
.await?
.into_iter()
.map(|row| row.name)
.collect()
};
let author = user::Entity::find_by_id(issue.author_id)
.one(self.db.read())
.await?;
let mut content = vec![issue.description.unwrap_or_default(), issue.body];
content.extend(labels);
content.extend(comments.into_iter().map(|row| row.content));
let document = SearchDocument {
id: issue.id.to_string(),
kind: SearchKind::Issue.as_str().to_string(),
space_id: Some(space.id.to_string()),
space_name: Some(space.name),
board_id: Some(board.id.to_string()),
number: Some(issue.autoincrement_id),
state: Some(
if issue.closed_at.is_some() {
"closed"
} else {
"open"
}
.to_string(),
),
title: issue.title,
content: format!(
"{}\n{}",
author.map(|row| row.username).unwrap_or_default(),
content.join("\n")
),
..Default::default()
};
meili
.add_documents(&self.index(SearchKind::Issue), &[document])
.await
}
pub async fn index_pull(&self, pull_id: Uuid) -> Result<(), SearchError> {
self.ensure_indexes().await?;
let meili = self.meili()?;
let Some(pull) = pull_requests::Entity::find_by_id(pull_id)
.one(self.db.read())
.await?
else {
return meili
.delete_document(&self.index(SearchKind::Pull), &pull_id.to_string())
.await;
};
let Some(context) = self.repo_context(pull.repo_id).await? else {
return Ok(());
};
let comments = pull_request_comment::Entity::find()
.filter(pull_request_comment::Column::PullRequestId.eq(pull.id))
.order_by_asc(pull_request_comment::Column::CreatedAt)
.all(self.db.read())
.await?;
let reviews = pull_request_review::Entity::find()
.filter(pull_request_review::Column::PullRequestId.eq(pull.id))
.order_by_asc(pull_request_review::Column::CreatedAt)
.all(self.db.read())
.await?;
let label_ids: Vec<Uuid> = pull_request_label::Entity::find()
.filter(pull_request_label::Column::PullRequestId.eq(pull.id))
.all(self.db.read())
.await?
.into_iter()
.map(|row| row.label_id)
.collect();
let labels = if label_ids.is_empty() {
Vec::new()
} else {
issue_label::Entity::find()
.filter(issue_label::Column::Id.is_in(label_ids))
.all(self.db.read())
.await?
.into_iter()
.map(|row| row.name)
.collect()
};
let author = user::Entity::find_by_id(pull.author_id)
.one(self.db.read())
.await?;
let mut content = vec![pull.body];
content.extend(labels);
content.extend(comments.into_iter().map(|row| row.body));
content.extend(reviews.into_iter().filter_map(|row| row.content));
let state = if pull.has_merged {
"merged"
} else if pull.closed_at.is_some() {
"closed"
} else {
"open"
};
let document = SearchDocument {
id: pull.id.to_string(),
kind: SearchKind::Pull.as_str().to_string(),
repo_id: Some(context.repo.id.to_string()),
space_id: Some(context.space.id.to_string()),
space_name: Some(context.space.name),
repo_name: Some(context.repo.name),
number: Some(pull.autoincrement_id),
state: Some(state.to_string()),
title: pull.title,
content: format!(
"{}\n{}",
author.map(|row| row.username).unwrap_or_default(),
content.join("\n")
),
..Default::default()
};
meili
.add_documents(&self.index(SearchKind::Pull), &[document])
.await
}
pub async fn index_message(&self, message_id: Uuid) -> Result<(), SearchError> {
self.ensure_indexes().await?;
let meili = self.meili()?;
let index = self.index(SearchKind::Message);
let Some(message) = messages::Entity::find_by_id(message_id)
.one(self.db.read())
.await?
else {
return meili.delete_document(&index, &message_id.to_string()).await;
};
let Some(document) = self.message_document(&message).await? else {
return meili.delete_document(&index, &message_id.to_string()).await;
};
meili.add_documents(&index, &[document]).await
}
/// 房间级回填：先清掉该房间的旧文档再重建，避免编辑/删除历史残留。
pub async fn index_room_messages(&self, room_id: Uuid) -> Result<(), SearchError> {
self.ensure_indexes().await?;
let meili = self.meili()?;
let index = self.index(SearchKind::Message);
let filter = format!("room_id = \"{room_id}\"");
meili.delete_filter(&index, &filter).await?;
let Some(room) = rooms::Entity::find_by_id(room_id)
.one(self.db.read())
.await?
else {
return Ok(());
};
if !room.is_active {
return Ok(());
}
let space = match room.space_id {
Some(space_id) => self.space_for_id(space_id).await?,
None => None,
};
// 从最新一条往旧分页，避免一次把大房间整表读进内存；作者按页批量取，
// 房间与空间只查一次，回填不会退化成逐条三查询。
let mut documents = Vec::new();
let mut capped = false;
let mut before_seq: Option<i64> = None;
loop {
let mut select = messages::Entity::find()
.filter(messages::Column::RoomId.eq(room_id))
.filter(messages::Column::DeletedAt.is_null())
.order_by_desc(messages::Column::Seq)
.limit(MESSAGE_BACKFILL_PAGE);
if let Some(seq) = before_seq {
select = select.filter(messages::Column::Seq.lt(seq));
}
let rows = select.all(self.db.read()).await?;
if rows.is_empty() {
break;
}
before_seq = rows.last().map(|message| message.seq);
let author_ids: Vec<i64> = rows.iter().map(|message| message.author_id).collect();
let authors = self.author_titles(&author_ids).await?;
for message in rows {
if documents.len() >= MAX_ROOM_MESSAGE_DOCUMENTS {
capped = true;
break;
}
if let Some(document) =
build_message_document(&message, &room, space.as_ref(), &authors)
{
documents.push(document);
}
}
if capped || before_seq.is_none() {
break;
}
}
if capped {
warn!(
%room_id,
limit = MAX_ROOM_MESSAGE_DOCUMENTS,
"search message index reached room document cap"
);
}
for batch in documents.chunks(self.config.batch_size) {
meili.add_documents(&index, batch).await?;
}
Ok(())
}
/// 单条消息的检索文档；房间/空间已失效或消息不可索引时返回 `None`。
async fn message_document(
&self,
message: &messages::Model,
) -> Result<Option<SearchDocument>, SearchError> {
let Some(room) = rooms::Entity::find_by_id(message.room_id)
.one(self.db.read())
.await?
else {
return Ok(None);
};
let space = match room.space_id {
Some(space_id) => self.space_for_id(space_id).await?,
None => None,
};
let authors = self.author_titles(&[message.author_id]).await?;
Ok(build_message_document(
message,
&room,
space.as_ref(),
&authors,
))
}
/// 消息作者在检索结果里的标题（显示名）。
async fn author_titles(
&self,
author_ids: &[i64],
) -> Result<HashMap<i64, String>, SearchError> {
if author_ids.is_empty() {
return Ok(HashMap::new());
}
let mut ids: Vec<i64> = author_ids.to_vec();
ids.sort();
ids.dedup();
Ok(user::Entity::find()
.filter(user::Column::Uid.is_in(ids))
.all(self.db.read())
.await?
.into_iter()
.map(|row| (row.uid, row.display_name))
.collect())
}
async fn space_for_id(&self, space_id: Uuid) -> Result<Option<spaces::Model>, SearchError> {
Ok(spaces::Entity::find_by_id(space_id)
.one(self.db.read())
.await?)
}
pub async fn queue_all(&self) -> Result<(), SearchError> {
crate::outbox::enqueue_all(self.db.write()).await?;
Ok(())
}
pub async fn queue_initial(&self) -> Result<(), SearchError> {
if !self.enabled() {
return Ok(());
}
// Several workers (and several processes) start at once; without a lock
// they all observe an empty queue and each schedule a full rebuild, which
// multiplies every index job. Serialize the check-and-enqueue with a
// transaction-scoped advisory lock so exactly one caller schedules it.
use sea_orm::{ConnectionTrait, DatabaseBackend, Statement, TransactionTrait};
let txn = self.db.primary().begin().await?;
txn.query_one_raw(Statement::from_string(
DatabaseBackend::Postgres,
"SELECT pg_advisory_xact_lock(721123)".to_string(),
))
.await?;
let has_work = models::search::index_job::Entity::find()
.filter(models::search::index_job::Column::State.eq("queued"))
.one(&txn)
.await?
.is_some();
if !has_work {
self.queue_all().await?;
}
txn.commit().await?;
Ok(())
}
async fn repo_context(&self, repo_id: Uuid) -> Result<Option<RepoContext>, SearchError> {
let Some(repo) = repos::Entity::find_by_id(repo_id)
.one(self.db.read())
.await?
else {
return Ok(None);
};
let Some(space) = self.space_for_dir(repo.dir_id).await? else {
return Ok(None);
};
Ok(Some(RepoContext { repo, space }))
}
async fn space_for_dir(&self, dir_id: Uuid) -> Result<Option<spaces::Model>, SearchError> {
let Some(dir) = dirs::Entity::find_by_id(dir_id).one(self.db.read()).await? else {
return Ok(None);
};
let Some(space_id) = dir.space_id else {
return Ok(None);
};
Ok(spaces::Entity::find_by_id(space_id)
.one(self.db.read())
.await?)
}
async fn visible_repos(&self, caller: Option<i64>) -> Result<Vec<RepoContext>, SearchError> {
let repos = repos::Entity::find()
.filter(repos::Column::IsArchived.eq(false))
.all(self.db.read())
.await?;
let dir_ids: Vec<Uuid> = repos.iter().map(|repo| repo.dir_id).collect();
let dirs = dirs::Entity::find()
.filter(dirs::Column::Id.is_in(dir_ids))
.all(self.db.read())
.await?;
let space_ids: Vec<Uuid> = dirs.iter().filter_map(|dir| dir.space_id).collect();
let spaces = spaces::Entity::find()
.filter(spaces::Column::Id.is_in(space_ids))
.all(self.db.read())
.await?;
let dirs: HashMap<Uuid, dirs::Model> = dirs.into_iter().map(|dir| (dir.id, dir)).collect();
let spaces: HashMap<Uuid, spaces::Model> =
spaces.into_iter().map(|space| (space.id, space)).collect();
let collaborations: HashSet<Uuid> = match caller {
Some(uid) => repo_collaborations::Entity::find()
.filter(repo_collaborations::Column::UserId.eq(uid))
.all(self.db.read())
.await?
.into_iter()
.map(|row| row.repo_id)
.collect(),
None => HashSet::new(),
};
Ok(repos
.into_iter()
.filter_map(|repo| {
let space = dirs
.get(&repo.dir_id)?
.space_id
.and_then(|id| spaces.get(&id))?
.clone();
let visible = !repo.is_private
|| caller.is_some_and(|uid| uid == space.owner_id)
|| collaborations.contains(&repo.id);
visible.then_some(RepoContext { repo, space })
})
.collect())
}
async fn visible_space_ids(&self, caller: Option<i64>) -> Result<Vec<String>, SearchError> {
let Some(caller) = caller else {
return Ok(Vec::new());
};
let owned = spaces::Entity::find()
.filter(spaces::Column::OwnerId.eq(caller))
.all(self.db.read())
.await?;
let member_ids: Vec<Uuid> = space_members::Entity::find()
.filter(space_members::Column::UserId.eq(caller))
.all(self.db.read())
.await?
.into_iter()
.map(|row| row.space_id)
.collect();
let mut ids: HashSet<String> = owned
.into_iter()
.map(|space| space.id.to_string())
.collect();
ids.extend(member_ids.into_iter().map(|id| id.to_string()));
Ok(ids.into_iter().collect())
}
/// 聊天消息的可见房间：来访者所在空间里的公共频道，加上他本人所在的
/// 私密频道与 DM/群聊；空间封禁与房间封禁都会把房间摘掉。
async fn visible_room_ids(&self, caller: Option<i64>) -> Result<Vec<String>, SearchError> {
let Some(caller) = caller else {
return Ok(Vec::new());
};
let spaces = spaces::Entity::find()
.filter(spaces::Column::IsActive.eq(true))
.all(self.db.read())
.await?;
let memberships = space_members::Entity::find()
.filter(space_members::Column::UserId.eq(caller))
.filter(space_members::Column::Status.eq(space_members::MemberStatus::Active))
.all(self.db.read())
.await?;
let member_of: HashSet<Uuid> = memberships.into_iter().map(|row| row.space_id).collect();
let visible_spaces: HashSet<Uuid> = spaces
.into_iter()
.filter(|space| space.owner_id == caller || member_of.contains(&space.id))
.map(|space| space.id)
.collect();
let rooms = rooms::Entity::find()
.filter(rooms::Column::IsActive.eq(true))
.all(self.db.read())
.await?;
let room_ids: Vec<Uuid> = rooms.iter().map(|room| room.id).collect();
let room_memberships = room_members::Entity::find()
.filter(room_members::Column::RoomId.is_in(room_ids.clone()))
.filter(room_members::Column::UserId.eq(caller))
.filter(room_members::Column::LeftAt.is_null())
.all(self.db.read())
.await?;
let joined_rooms: HashSet<Uuid> = room_memberships
.into_iter()
.map(|row| row.room_id)
.collect();
let bans = room_bans::Entity::find()
.filter(room_bans::Column::RoomId.is_in(room_ids))
.filter(room_bans::Column::UserId.eq(caller))
.all(self.db.read())
.await?;
let now = time::OffsetDateTime::now_utc();
let banned: HashSet<Uuid> = bans
.into_iter()
.filter(|row| row.expires_at.is_none_or(|expires| expires > now))
.map(|row| row.room_id)
.collect();
let visible = rooms
.into_iter()
.filter(|room| {
if banned.contains(&room.id) {
return false;
}
match room.space_id {
Some(space_id) => {
visible_spaces.contains(&space_id)
&& (!room.is_private
|| room.owner_id == caller
|| joined_rooms.contains(&room.id))
}
None => joined_rooms.contains(&room.id),
}
})
.map(|room| room.id.to_string())
.collect();
Ok(visible)
}
}
/// 组装一条消息的检索文档；房间/空间失效、系统/临时消息或空正文都不索引。
/// 返回 `None` 意味着该消息应当从消息索引里删除。
fn build_message_document(
message: &messages::Model,
room: &rooms::Model,
space: Option<&spaces::Model>,
authors: &HashMap<i64, String>,
) -> Option<SearchDocument> {
if message.deleted_at.is_some() {
return None;
}
if message.flags & FLAG_EPHEMERAL != 0 {
return None;
}
if !matches!(
message.message_type,
ChatMessageType::Default | ChatMessageType::Reply | ChatMessageType::ThreadCreated
) {
return None;
}
if message.content.trim().is_empty() {
return None;
}
if !room.is_active {
return None;
}
if room.space_id.is_some() && space.is_none_or(|space| !space.is_active) {
return None;
}
Some(SearchDocument {
id: message.id.to_string(),
kind: SearchKind::Message.as_str().to_string(),
space_id: space.map(|space| space.id.to_string()),
space_name: space.map(|space| space.name.clone()),
room_id: Some(room.id.to_string()),
room_name: Some(room.name.clone()),
thread_id: message.thread_id.map(|id| id.to_string()),
title: authors
.get(&message.author_id)
.cloned()
.unwrap_or_else(|| message.author_id.to_string()),
content: message.content.clone(),
..Default::default()
})
}
/// 各类型按相关度轮流出结果，避免符号命中把代码命中挤出结果页。
fn merge_buckets(buckets: Vec<Vec<SearchHit>>, limit: usize) -> Vec<SearchHit> {
let mut merged: Vec<SearchHit> = Vec::with_capacity(limit);
let mut rank = 0usize;
while merged.len() < limit {
let mut progressed = false;
for bucket in &buckets {
let Some(hit) = bucket.get(rank) else {
continue;
};
merged.push(hit.clone());
progressed = true;
if merged.len() == limit {
break;
}
}
if !progressed {
break;
}
rank += 1;
}
merged
}
struct FilterOptions<'a> {
definitions_only: bool,
repo_ids: &'a [String],
space_ids: &'a [String],
room_ids: &'a [String],
space_name: Option<&'a str>,
repo_name: Option<&'a str>,
board_id: Option<String>,
room_id: Option<String>,
language: Option<&'a str>,
state: Option<&'a str>,
}
fn build_filter(kind: SearchKind, options: &FilterOptions<'_>) -> Option<String> {
let mut filters = Vec::new();
if matches!(
kind,
SearchKind::Code | SearchKind::Symbol | SearchKind::Pull
) {
filters.push(in_filter("repo_id", options.repo_ids));
} else if kind == SearchKind::Message {
filters.push(in_filter("room_id", options.room_ids));
} else {
filters.push(in_filter("space_id", options.space_ids));
}
if kind == SearchKind::Symbol && options.definitions_only {
filters.push("definition = true".to_string());
}
for (field, value) in [
("space_name", options.space_name),
("repo_name", options.repo_name),
("language", options.language),
("state", options.state),
] {
if let Some(value) = value.filter(|value| !value.trim().is_empty()) {
filters.push(eq_filter(field, value));
}
}
// 看板与频道是「当前所在位置」的默认约束，只在对应类型上生效：工单按 board_id，
// 消息按 room_id。其余类型的文档没有这两个字段，加了条件会命中零条。
if kind == SearchKind::Issue
&& let Some(board) = options
.board_id
.as_deref()
.filter(|value| !value.is_empty())
{
filters.push(eq_filter("board_id", board));
}
if kind == SearchKind::Message
&& let Some(room) = options.room_id.as_deref().filter(|value| !value.is_empty())
{
filters.push(eq_filter("room_id", room));
}
Some(filters.join(" AND "))
}
fn in_filter(field: &str, values: &[String]) -> String {
let values = values
.iter()
.map(|value| format!("\"{}\"", escape_filter(value)))
.collect::<Vec<_>>()
.join(", ");
format!("{field} IN [{values}]")
}
fn eq_filter(field: &str, value: &str) -> String {
format!("{field} = \"{}\"", escape_filter(value))
}
fn escape_filter(value: &str) -> String {
value.replace('\\', "\\\\").replace('"', "\\\"")
}
/// 符号索引里“引用”条目的上限：一个文件的标识符出现次数可以非常多，不设上限时
/// 大仓库会把索引撑爆并把定义挤出结果。
const MAX_REFERENCE_DOCUMENTS: usize = 4_000;
/// 单个仓库代码分块的上限，防止超大仓库一次索引把检索服务打满。
const MAX_CODE_DOCUMENTS: usize = 20_000;
/// 单个房间回填的消息上限（按 seq 取最新），防止超大频道把检索服务打满；
/// 增量消息不设上限，回填只是历史补齐。
const MAX_ROOM_MESSAGE_DOCUMENTS: usize = 50_000;
/// 回填时每次从数据库翻多少行，避免整房间一次性读进内存。
const MESSAGE_BACKFILL_PAGE: u64 = 500;
/// 增量删除路径时每批 filter 携带的路径数，避免 Meili 请求体过大。
const DELETE_PATH_CHUNK: usize = 200;
#[derive(Clone)]
struct DocContext {
repo_id: Uuid,
space_id: Uuid,
space_name: String,
repo_name: String,
sha: String,
chunk_lines: usize,
}
fn parse_file(
context: &DocContext,
path: &str,
language: &'static str,
text: &str,
) -> (Vec<SearchDocument>, Vec<SearchDocument>) {
let mut code_documents = Vec::new();
for (start, end, chunk) in text_chunks(text, context.chunk_lines) {
code_documents.push(SearchDocument {
id: Uuid::now_v7().to_string(),
kind: SearchKind::Code.as_str().to_string(),
repo_id: Some(context.repo_id.to_string()),
space_id: Some(context.space_id.to_string()),
space_name: Some(context.space_name.clone()),
repo_name: Some(context.repo_name.clone()),
revision: Some(context.sha.clone()),
path: Some(path.to_string()),
line_start: Some(start),
line_end: Some(end),
language: Some(language.to_string()),
title: path.to_string(),
content: chunk,
..Default::default()
});
}
let mut symbol_documents = Vec::new();
for (ordinal, symbol) in symbols_for(path, text).into_iter().enumerate() {
symbol_documents.push(SearchDocument {
id: Uuid::now_v7().to_string(),
kind: SearchKind::Symbol.as_str().to_string(),
repo_id: Some(context.repo_id.to_string()),
space_id: Some(context.space_id.to_string()),
space_name: Some(context.space_name.clone()),
repo_name: Some(context.repo_name.clone()),
revision: Some(context.sha.clone()),
path: Some(path.to_string()),
line_start: Some(symbol.line),
line_end: Some(symbol.end_line),
language: Some(language.to_string()),
symbol_id: Some(format!(
"{}:{}:{}:{}:{ordinal}",
context.repo_id, context.sha, path, symbol.line
)),
symbol_name: Some(symbol.name.clone()),
symbol_kind: Some(symbol.kind.clone()),
definition: Some(symbol.is_definition),
title: symbol.name,
content: format!("{}:{}", path, symbol.line),
..Default::default()
});
}
(code_documents, symbol_documents)
}
struct IndexSink<'a> {
meili: &'a MeiliClient,
code_index: &'a str,
symbol_index: &'a str,
batch_size: usize,
pending_code: Vec<SearchDocument>,
pending_symbols: Vec<SearchDocument>,
uids: Vec<u64>,
code_count: usize,
reference_count: usize,
symbol_count: usize,
files: usize,
capped: bool,
}
impl<'a> IndexSink<'a> {
fn new(
meili: &'a MeiliClient,
code_index: &'a str,
symbol_index: &'a str,
batch_size: usize,
) -> Self {
Self {
meili,
code_index,
symbol_index,
batch_size,
pending_code: Vec::new(),
pending_symbols: Vec::new(),
uids: Vec::new(),
code_count: 0,
reference_count: 0,
symbol_count: 0,
files: 0,
capped: false,
}
}
async fn absorb(
&mut self,
code_docs: Vec<SearchDocument>,
symbol_docs: Vec<SearchDocument>,
) -> Result<(), SearchError> {
self.files += 1;
for document in code_docs {
if self.code_count >= MAX_CODE_DOCUMENTS {
self.capped = true;
break;
}
self.code_count += 1;
self.pending_code.push(document);
}
for document in symbol_docs {
if document.definition == Some(true) {
self.symbol_count += 1;
self.pending_symbols.push(document);
} else if self.reference_count < MAX_REFERENCE_DOCUMENTS {
self.reference_count += 1;
self.symbol_count += 1;
self.pending_symbols.push(document);
}
}
while self.pending_code.len() >= self.batch_size {
let batch: Vec<SearchDocument> = self.pending_code.drain(..self.batch_size).collect();
if let Some(uid) = self.meili.submit_documents(self.code_index, &batch).await? {
self.uids.push(uid);
}
}
while self.pending_symbols.len() >= self.batch_size {
let batch: Vec<SearchDocument> =
self.pending_symbols.drain(..self.batch_size).collect();
if let Some(uid) = self
.meili
.submit_documents(self.symbol_index, &batch)
.await?
{
self.uids.push(uid);
}
}
Ok(())
}
async fn finish(&mut self) -> Result<(), SearchError> {
if !self.pending_code.is_empty() {
let batch = std::mem::take(&mut self.pending_code);
if let Some(uid) = self.meili.submit_documents(self.code_index, &batch).await? {
self.uids.push(uid);
}
}
if !self.pending_symbols.is_empty() {
let batch = std::mem::take(&mut self.pending_symbols);
if let Some(uid) = self
.meili
.submit_documents(self.symbol_index, &batch)
.await?
{
self.uids.push(uid);
}
}
Ok(())
}
fn files(&self) -> usize {
self.files
}
fn code_count(&self) -> usize {
self.code_count
}
fn symbol_count(&self) -> usize {
self.symbol_count
}
fn capped(&self) -> bool {
self.capped
}
}
fn path_filter(repo_id: Uuid, paths: &[String]) -> String {
let list = paths
.iter()
.map(|path| format!("\"{}\"", escape_filter(path)))
.collect::<Vec<_>>()
.join(", ");
format!("repo_id = \"{repo_id}\" AND path IN [{list}]")
}
fn join_error(error: tokio::task::JoinError) -> SearchError {
SearchError::Git(format!("search task join failed: {error}"))
}
fn compact_snippet(value: &str) -> String {
let normalized = value.split_whitespace().collect::<Vec<_>>().join(" ");
let mut chars = normalized.chars();
let snippet: String = chars.by_ref().take(280).collect();
if chars.next().is_some() {
format!("{snippet}…")
} else {
snippet
}
}