feat(search): add initial implementation for search service with indexing and API endpoints

Modified app/synchost/src/main.rs +14 −2 View app/synchost/src/main.rs

@@ -6,6 +6,7 @@

66use cache::CacheClient;

77use config::scoped::{ScopedConfig, SynchostConfig};

88use git::sync::{NoopWebhookTx, SyncQueue, SyncWorker, WebhookTx};

9use search::{SearchService, SearchWorker};

910use tokio::net::TcpListener;

1011use tokio_util::sync::CancellationToken;

1112use tracing::info;

@@ -35,6 +36,19 @@

3536

3637 let cancel = CancellationToken::new();

3738

39 let search = SearchService::from_parts(db.clone(), cfg.search.clone())

40 .map_err(|e| anyhow::anyhow!("building search service: {e}"))?;

41 if search.enabled() {

42 let worker = SearchWorker::new(

43 db.clone(),

44 search,

45 PathBuf::from(cfg.git.repo_root.clone()),

46 cancel.clone(),

47 );

48 tokio::spawn(worker.run());

49 info!("search index worker started");

50 }

51

3852 if cfg.health.enabled() {

3953 let health_addr: SocketAddr =

4054 format!("{}:{}", cfg.health.bind_addr(), cfg.health.synchost_port())

@@ -89,5 +103,3 @@

89103 info!("synchost shut down");

90104 Ok(())

91105}

92

93

Modified app/synchost/Cargo.toml +2 −1 View app/synchost/Cargo.toml

@@ -9,9 +9,10 @@

99export = { workspace = true }

1010config = { workspace = true }

1111cache = { workspace = true }

12search = { workspace = true }

1213axum = { workspace = true }

1314tokio = { workspace = true, features = ["signal", "net", "rt-multi-thread", "macros"] }

1415tokio-util = { workspace = true }

1516tracing = { workspace = true }

1617tracing-subscriber = { workspace = true }

17anyhow = { workspace = true }

No newline at end of file

18anyhow = { workspace = true }

Modified chart/values.yaml +4 −0 View chart/values.yaml

@@ -32,6 +32,9 @@

3232 # APP_DATABASE__REPLICAS: '[{url="postgres://readonly@pg:5432/gitdata"}]'

3333 APP_EMAIL__ENABLED: "false"

3434 APP_EMAIL_QUEUE__ENABLED: "false"

35 APP_SEARCH__ENABLED: "false"

36 APP_SEARCH__URL: "http://meilisearch:7700"

37 APP_SEARCH__INDEX_PREFIX: "gitdata"

3538

3639# 敏感配置（渲染为 Secret）。注意 AppConfig 的 Debug 会带出密钥，

3740# 切勿在任何日志里 {:?} 打印配置。

@@ -48,6 +51,7 @@

4851 APP_S3__SECRET_ACCESS_KEY: ""

4952 APP_EMAIL__SMTP_PASSWORD: ""

5053 APP_AUTH__CHAT_JWT_SEC: "" # >=32 字节，chat WS 握手 JWT 密钥

54 APP_SEARCH__API_KEY: ""

5155

5256podSecurityContext:

5357 fsGroup: 10001

Modified config/default.toml +11 −0 View config/default.toml

@@ -32,6 +32,17 @@

3232username = ""

3333password = ""

3434

35[search]

36enabled = false

37url = "http://127.0.0.1:7700"

38api_key = ""

39index_prefix = "gitdata"

40max_file_bytes = 1048576

41chunk_lines = 160

42batch_size = 100

43poll_interval_secs = 2

44task_timeout_secs = 120

45

3546[database]

3647url = "postgres://postgres:postgres@localhost:5432/gitdata"

3748host = ""

Modified lib/config/src/lib.rs +5 −1 View lib/config/src/lib.rs

@@ -15,6 +15,7 @@

1515pub mod redis;

1616pub mod s3;

1717pub mod scoped;

18pub mod search;

1819pub mod session;

1920pub mod space;

2021pub mod user;

@@ -34,12 +35,13 @@

3435use crate::email_queue::EmailQueueConfig;

3536use crate::git::GitConfig;

3637use crate::githost::GithostConfig;

37use crate::im::ImServerConfig;

3838use crate::health::HealthConfig;

39use crate::im::ImServerConfig;

3940use crate::issues::IssueConfig;

4041use crate::pow::PowConfig;

4142use crate::redis::RedisConfig;

4243use crate::s3::S3Config;

44use crate::search::SearchConfig;

4345use crate::session::SessionConfig;

4446use crate::space::SpaceConfig;

4547use crate::user::UserConfig;

@@ -55,6 +57,7 @@

5557 pub chat: ChatConfig,

5658 pub database: DatabaseConfig,

5759 pub session: SessionConfig,

60 pub search: SearchConfig,

5861 pub pow: PowConfig,

5962 pub auth: AuthConfig,

6063 pub web: WebConfig,

@@ -88,6 +91,7 @@

8891 .is_valid()

8992 .map_err(ConfigError::Validation)?;

9093 self.session.is_valid().map_err(ConfigError::Validation)?;

94 self.search.validate().map_err(ConfigError::Validation)?;

9195 self.apihost.is_valid().map_err(ConfigError::Validation)?;

9296 self.githost.is_valid().map_err(ConfigError::Validation)?;

9397 self.im.is_valid().map_err(ConfigError::Validation)?;

Modified lib/config/src/scoped.rs +6 −0 View lib/config/src/scoped.rs

@@ -18,6 +18,7 @@

1818use crate::quota::QuotaConfig;

1919use crate::redis::RedisConfig;

2020use crate::s3::S3Config;

21use crate::search::SearchConfig;

2122use crate::session::SessionConfig;

2223use crate::space::SpaceConfig;

2324use crate::user::UserConfig;

@@ -72,6 +73,7 @@

7273 pub cache: CacheConfig,

7374 pub git: GitConfig,

7475 pub health: HealthConfig,

76 pub search: SearchConfig,

7577}

7678

7779impl ScopedConfig for SynchostConfig {

@@ -79,6 +81,7 @@

7981 self.database.validate()?;

8082 self.redis.validate()?;

8183 self.cache.validate()?;

84 self.search.validate()?;

8285 self.health.is_valid()

8386 }

8487}

@@ -145,6 +148,7 @@

145148 pub email: EmailConfig,

146149 pub email_queue: EmailQueueConfig,

147150 pub apihost: ApihostSection,

151 pub search: SearchConfig,

148152}

149153

150154impl ApihostConfig {

@@ -167,6 +171,7 @@

167171 email: self.email,

168172 email_queue: self.email_queue,

169173 apihost: self.apihost,

174 search: self.search,

170175 ..Default::default()

171176 }

172177 }

@@ -183,6 +188,7 @@

183188 self.email.is_valid()?;

184189 self.email_queue.is_valid()?;

185190 self.session.is_valid()?;

191 self.search.validate()?;

186192 self.apihost.is_valid()

187193 }

188194}

Added lib/config/src/search.rs +55 −0 View lib/config/src/search.rs

@@ -0,0 +1,55 @@

1use serde::Deserialize;

2

3#[derive(Clone, Debug, Deserialize)]

4#[serde(default)]

5pub struct SearchConfig {

6 pub enabled: bool,

7 pub url: String,

8 pub api_key: String,

9 pub index_prefix: String,

10 pub max_file_bytes: usize,

11 pub chunk_lines: usize,

12 pub batch_size: usize,

13 pub poll_interval_secs: u64,

14 pub task_timeout_secs: u64,

15}

16

17impl Default for SearchConfig {

18 fn default() -> Self {

19 Self {

20 enabled: false,

21 url: "http://127.0.0.1:7700".to_string(),

22 api_key: String::new(),

23 index_prefix: "gitdata".to_string(),

24 max_file_bytes: 1024 * 1024,

25 chunk_lines: 160,

26 batch_size: 100,

27 poll_interval_secs: 2,

28 task_timeout_secs: 120,

29 }

30 }

31}

32

33impl SearchConfig {

34 pub fn validate(&self) -> Result<(), String> {

35 if !self.enabled {

36 return Ok(());

37 }

38 if !(self.url.starts_with("http://") || self.url.starts_with("https://")) {

39 return Err("search.url must be an http(s) URL".to_string());

40 }

41 if self.api_key.trim().is_empty() {

42 return Err("search.api_key must not be empty when search is enabled".to_string());

43 }

44 if self.index_prefix.trim().is_empty() {

45 return Err("search.index_prefix must not be empty".to_string());

46 }

47 if self.max_file_bytes == 0 || self.chunk_lines == 0 || self.batch_size == 0 {

48 return Err("search limits must be greater than zero".to_string());

49 }

50 if self.poll_interval_secs == 0 || self.task_timeout_secs == 0 {

51 return Err("search intervals must be greater than zero".to_string());

52 }

53 Ok(())

54 }

55}

Modified lib/git/src/sync/sync.rs +4 −0 View lib/git/src/sync/sync.rs

@@ -169,6 +169,10 @@

169169 license::sync_repo_license(&txn, &self.repo_dir(), &self.repo).await?;

170170 tracing::debug!(repo_id = %self.repo.id, "license synced");

171171

172 search::outbox::enqueue_repo(&txn, self.repo.id)

173 .await

174 .map_err(SyncError::Db)?;

175

172176 txn.commit().await.map_err(SyncError::Db)?;

173177 tracing::debug!(repo_id = %self.repo.id, "sync transaction committed");

174178

Modified lib/git/Cargo.toml +2 −1 View lib/git/Cargo.toml

@@ -35,6 +35,7 @@

3535rand = { workspace = true }

3636bytes = { workspace = true }

3737session = { workspace = true }

38search = { workspace = true }

3839utoipa = { workspace = true }

3940maud = { workspace = true }

40pulldown-cmark = { workspace = true }

No newline at end of file

41pulldown-cmark = { workspace = true }

Modified lib/issues/src/services/comments.rs +14 −3 View lib/issues/src/services/comments.rs

@@ -1,6 +1,6 @@

1use models::issue::issue_event_line::IssueEventType;

21use models::issue::comment;

32use models::issue::issue_content_history;

3use models::issue::issue_event_line::IssueEventType;

44use models::issue::issue_member::{self, IssueMemberKind};

55use models::users::user_notifications::NotificationType;

66use sea_orm::sea_query::Expr;

@@ -90,7 +90,8 @@

9090 "comment creation quota exceeded, retry tomorrow".to_string(),

9191 ));

9292 }

93 let (space, board, item) = issue_context(self, caller, space_name, board_id, issue_id).await?;

93 let (space, board, item) =

94 issue_context(self, caller, space_name, board_id, issue_id).await?;

9495

9596 let content = cmd.content.trim();

9697 if content.is_empty() || content.chars().count() > self.config().issues.max_comment_len {

@@ -281,6 +282,8 @@

281282 let user_map = self.load_user_briefs(&[inserted.poster_id]).await?;

282283 let poster = user_map.get(&inserted.poster_id).cloned();

283284

285 self.queue_search_issue(issue_id).await?;

286

284287 Ok(CommentDto {

285288 id: inserted.id,

286289 issue_id: inserted.issue_id,

@@ -343,7 +346,12 @@

343346 })

344347 .collect();

345348

346 Ok(CommentPage { items, total, limit, offset })

349 Ok(CommentPage {

350 items,

351 total,

352 limit,

353 offset,

354 })

347355 }

348356

349357 pub async fn list_comments_tree(

@@ -509,6 +517,8 @@

509517 let user_map = self.load_user_briefs(&[updated.poster_id]).await?;

510518 let poster = user_map.get(&updated.poster_id).cloned();

511519

520 self.queue_search_issue(issue_id).await?;

521

512522 Ok(CommentDto {

513523 id: updated.id,

514524 issue_id: updated.issue_id,

@@ -569,6 +579,7 @@

569579 })

570580 .await?;

571581

582 self.queue_search_issue(issue_id).await?;

572583 Ok(())

573584 }

574585}

Modified lib/issues/src/services/issues.rs +26 −10 View lib/issues/src/services/issues.rs

@@ -1,12 +1,11 @@

1use models::issue::issue_item::IssuePriority;

2use models::issue::issue_label;

3use models::space::space_members::{self, MemberStatus};

4use models::issue::issue_event_line::IssueEventType;

51use models::issue::comment;

62use models::issue::issue_content_history;

73use models::issue::issue_dependency;

84use models::issue::issue_event_line;

5use models::issue::issue_event_line::IssueEventType;

96use models::issue::issue_item;

7use models::issue::issue_item::IssuePriority;

8use models::issue::issue_label;

109use models::issue::issue_member;

1110use models::issue::issue_pin;

1211use models::issue::issue_reactions;

@@ -19,6 +18,7 @@

1918use models::issue::milestone;

2019use models::issue::stopwatch;

2120use models::issue::tracked_time;

21use models::space::space_members::{self, MemberStatus};

2222use models::users::user_notifications::NotificationType;

2323use sea_orm::{

2424 ActiveModelTrait, ColumnTrait, Condition, ConnectionTrait, EntityTrait, PaginatorTrait,

@@ -256,7 +256,9 @@

256256 let user_map = self.load_user_briefs(&uids).await?;

257257 for r in rows {

258258 if let Some(brief) = user_map.get(&r.user_id) {

259 map.entry(r.issue_id).or_insert_with(Vec::new).push(brief.clone());

259 map.entry(r.issue_id)

260 .or_insert_with(Vec::new)

261 .push(brief.clone());

260262 }

261263 }

262264 Ok(map)

@@ -539,6 +541,7 @@

539541 }

540542 }

541543

544 self.queue_search_issue(inserted.id).await?;

542545 self.build_issue_dto(caller, inserted).await

543546 }

544547

@@ -683,7 +686,8 @@

683686 issue_id: Uuid,

684687 cmd: UpdateIssueCmd,

685688 ) -> Result<IssueDto, IssueError> {

686 let (space, board, item) = issue_context(self, caller, space_name, board_id, issue_id).await?;

689 let (space, board, item) =

690 issue_context(self, caller, space_name, board_id, issue_id).await?;

687691

688692 if item.author_id != caller && !is_board_manager(self, &space, caller).await? {

689693 return Err(IssueError::Forbidden);

@@ -810,7 +814,10 @@

810814 if let Some(new_assignees) = assignee_replace {

811815 issue_member::Entity::delete_many()

812816 .filter(issue_member::Column::IssueId.eq(u.id))

813 .filter(issue_member::Column::Kind.eq(issue_member::IssueMemberKind::Assignee))

817 .filter(

818 issue_member::Column::Kind

819 .eq(issue_member::IssueMemberKind::Assignee),

820 )

814821 .exec(txn)

815822 .await?;

816823 for a_uid in new_assignees {

@@ -948,9 +955,9 @@

948955 })

949956 .await?;

950957

958 self.queue_search_issue(updated.id).await?;

951959 self.build_issue_dto(caller, updated).await

952960 }

953

954961}

955962

956963#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, ToSchema)]

@@ -1006,7 +1013,10 @@

10061013 issue_item::Column::ClosedAt,

10071014 sea_orm::sea_query::Expr::value(should_close.then_some(now)),

10081015 )

1009 .col_expr(issue_item::Column::UpdatedAt, sea_orm::sea_query::Expr::value(now))

1016 .col_expr(

1017 issue_item::Column::UpdatedAt,

1018 sea_orm::sea_query::Expr::value(now),

1019 )

10101020 .filter(issue_item::Column::Id.eq(item.id))

10111021 .exec(self.db().write())

10121022 .await?;

@@ -1055,6 +1065,7 @@

10551065 })

10561066 .await?;

10571067

1068 self.queue_search_issue(issue_id).await?;

10581069 Ok(())

10591070 }

10601071

@@ -1082,7 +1093,11 @@

10821093 am.closed_at = Set(closed.then_some(now));

10831094 am.updated_at = Set(Some(now));

10841095

1085 let event_type = if closed { IssueEventType::Closed } else { IssueEventType::Reopened };

1096 let event_type = if closed {

1097 IssueEventType::Closed

1098 } else {

1099 IssueEventType::Reopened

1100 };

10861101 let this = self.clone();

10871102

10881103 let updated = self

@@ -1098,6 +1113,7 @@

10981113 })

10991114 .await?;

11001115

1116 self.queue_search_issue(updated.id).await?;

11011117 self.build_issue_dto(caller, updated).await

11021118 }

11031119}

Modified lib/issues/src/services/labels.rs +30 −2 View lib/issues/src/services/labels.rs

@@ -13,7 +13,9 @@

1313use crate::errors::IssueError;

1414use crate::services::service::IssueService;

1515use crate::share::authz::require_board_manage;

16use crate::share::lookup::{board_context, lookup_board, lookup_issue, require_issue_access, space_context};

16use crate::share::lookup::{

17 board_context, lookup_board, lookup_issue, require_issue_access, space_context,

18};

1719

1820#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]

1921pub struct LabelDto {

@@ -220,7 +222,12 @@

220222 .await?;

221223

222224 let items = rows.into_iter().map(LabelDto::from).collect();

223 Ok(LabelPage { items, total, limit, offset })

225 Ok(LabelPage {

226 items,

227 total,

228 limit,

229 offset,

230 })

224231 }

225232

226233 pub async fn get_label(

@@ -301,6 +308,16 @@

301308 am.updated_at = Set(Some(now));

302309

303310 let updated = am.update(self.db().write()).await?;

311 let issue_ids: Vec<Uuid> = issue_relation_label::Entity::find()

312 .filter(issue_relation_label::Column::LabelId.eq(label_id))

313 .all(self.db().read())

314 .await?

315 .into_iter()

316 .map(|row| row.issue_id)

317 .collect();

318 for issue_id in issue_ids {

319 self.queue_search_issue(issue_id).await?;

320 }

304321 Ok(LabelDto::from(updated))

305322 }

306323

@@ -318,6 +335,13 @@

318335 .one(self.db().read())

319336 .await?

320337 .ok_or(IssueError::NotFound)?;

338 let issue_ids: Vec<Uuid> = issue_relation_label::Entity::find()

339 .filter(issue_relation_label::Column::LabelId.eq(label_id))

340 .all(self.db().read())

341 .await?

342 .into_iter()

343 .map(|row| row.issue_id)

344 .collect();

321345

322346 self.db()

323347 .primary()

@@ -335,6 +359,9 @@

335359 })

336360 .await?;

337361

362 for issue_id in issue_ids {

363 self.queue_search_issue(issue_id).await?;

364 }

338365 Ok(())

339366 }

340367

@@ -423,6 +450,7 @@

423450 })

424451 .await?;

425452

453 self.queue_search_issue(issue_id).await?;

426454 Ok(())

427455 }

428456

Modified lib/issues/src/services/service.rs +10 −4 View lib/issues/src/services/service.rs

@@ -47,6 +47,11 @@

4747 &self.state.database

4848 }

4949

50 pub async fn queue_search_issue(&self, issue_id: Uuid) -> Result<(), IssueError> {

51 search::outbox::enqueue_issue(self.db().write(), issue_id).await?;

52 Ok(())

53 }

54

5055 pub fn email(&self) -> &EmailService {

5156 &self.state.email

5257 }

@@ -83,9 +88,11 @@

8388 let space = crate::share::lookup::lookup_space_by_name(self, space_name).await?;

8489 crate::share::lookup::require_member(self, &space, caller).await?;

8590 let board = crate::share::lookup::lookup_board(self, &space, board_id, caller).await?;

86 Ok(crate::share::lookup::lookup_issue_by_ref(self, &board, issue_ref)

87 .await?

88 .id)

91 Ok(

92 crate::share::lookup::lookup_issue_by_ref(self, &board, issue_ref)

93 .await?

94 .id,

95 )

8996 }

9097

9198 pub(crate) async fn write_event(

@@ -221,7 +228,6 @@

221228 models::users::queries::primary_emails(self.db().read(), uids)

222229 .await

223230 .map_err(IssueError::from)

224

225231 }

226232

227233 pub(crate) async fn send_email(

Modified lib/issues/Cargo.toml +2 −1 View lib/issues/Cargo.toml

@@ -18,4 +18,5 @@

1818utoipa = { workspace = true }

1919uuid = { workspace = true }

2020time = { workspace = true }

21tracing = { workspace = true }

No newline at end of file

21tracing = { workspace = true }

22search = { workspace = true }

Added lib/models/src/search/index_job.rs +27 −0 View lib/models/src/search/index_job.rs

@@ -0,0 +1,27 @@

1use sea_orm::entity::prelude::*;

2use time::OffsetDateTime;

3use uuid::Uuid;

4

5#[sea_orm::model]

6#[derive(Clone, Debug, PartialEq, Eq, DeriveEntityModel)]

7#[sea_orm(table_name = "search_index_job")]

8pub struct Model {

9 #[sea_orm(primary_key, auto_increment = false)]

10 pub id: Uuid,

11 #[sea_orm(indexed)]

12 pub kind: String,

13 #[sea_orm(indexed)]

14 pub target_id: Option<Uuid>,

15 #[sea_orm(indexed)]

16 pub repo_id: Option<Uuid>,

17 #[sea_orm(indexed)]

18 pub state: String,

19 pub attempts: i32,

20 pub run_after: OffsetDateTime,

21 pub locked_at: Option<OffsetDateTime>,

22 pub last_error: Option<String>,

23 pub created_at: OffsetDateTime,

24 pub updated_at: OffsetDateTime,

25}

26

27impl ActiveModelBehavior for ActiveModel {}

Added lib/models/src/search/mod.rs +1 −0 View lib/models/src/search/mod.rs

@@ -0,0 +1 @@

1pub mod index_job;

Modified lib/models/src/lib.rs +1 −0 View lib/models/src/lib.rs

@@ -8,6 +8,7 @@

88pub mod pull_request;

99pub mod repo;

1010pub mod room;

11pub mod search;

1112pub mod space;

1213pub mod system;

1314pub mod users;

Modified lib/pulls/src/services/comments.rs +4 −0 View lib/pulls/src/services/comments.rs

@@ -170,6 +170,7 @@

170170 .await?;

171171 self.record_event(pull, Some(caller), EVENT_COMMENTED, commit_sha, None)

172172 .await?;

173 self.queue_search_pull(pull.id, pull.repo_id).await?;

173174 let _ = ctx;

174175 Ok(row)

175176 }

@@ -258,6 +259,7 @@

258259 model.body = Set(body);

259260 model.updated_at = Set(Some(OffsetDateTime::now_utc()));

260261 let updated = model.update(self.db_write()).await?;

262 self.queue_search_pull(pull.id, pull.repo_id).await?;

261263 let _ = pull;

262264 Ok(updated)

263265 }

@@ -281,6 +283,7 @@

281283 )

282284 .exec(self.db_write())

283285 .await?;

286 self.queue_search_pull(pull.id, pull.repo_id).await?;

284287 Ok(())

285288 }

286289

@@ -305,6 +308,7 @@

305308 });

306309 model.resolved_by = Set(if resolved { Some(caller) } else { None });

307310 let updated = model.update(self.db_write()).await?;

311 self.queue_search_pull(pull.id, pull.repo_id).await?;

308312 let _ = pull;

309313 Ok(updated)

310314 }

Modified lib/pulls/src/services/labels.rs +2 −0 View lib/pulls/src/services/labels.rs

@@ -96,6 +96,7 @@

9696 )

9797 .await?;

9898 }

99 self.queue_search_pull(pull.id, pull.repo_id).await?;

99100 self.list_labels(pull.id).await

100101 }

101102

@@ -123,6 +124,7 @@

123124 )

124125 .await?;

125126 }

127 self.queue_search_pull(pull.id, pull.repo_id).await?;

126128 self.list_labels(pull.id).await

127129 }

128130}

Modified lib/pulls/src/services/pulls.rs +6 −0 View lib/pulls/src/services/pulls.rs

@@ -409,6 +409,7 @@

409409 .await?;

410410 txn.commit().await?;

411411 self.recount_open(ctx.model.id).await?;

412 self.queue_search_pull(pull.id, pull.repo_id).await?;

412413 Ok(pull)

413414 }

414415

@@ -493,6 +494,7 @@

493494 let updated = model.update(self.db_write()).await?;

494495 self.record_event(&updated, Some(caller), EVENT_UPDATED, None, None)

495496 .await?;

497 self.queue_search_pull(updated.id, updated.repo_id).await?;

496498 let _ = (space_name, repo_name);

497499 Ok(updated)

498500 }

@@ -531,6 +533,7 @@

531533 )

532534 .await?;

533535 self.recount_open(updated.repo_id).await?;

536 self.queue_search_pull(updated.id, updated.repo_id).await?;

534537 Ok(updated)

535538 }

536539

@@ -576,6 +579,7 @@

576579 .await?;

577580 txn.commit().await?;

578581 self.recount_open(pull.repo_id).await?;

582 self.queue_search_pull(pull.id, pull.repo_id).await?;

579583 Ok(())

580584 }

581585

@@ -666,6 +670,7 @@

666670 self.record_event(&updated, Some(caller), EVENT_MERGED, Some(merged_sha), None)

667671 .await?;

668672 self.recount_open(updated.repo_id).await?;

673 self.queue_search_pull(updated.id, updated.repo_id).await?;

669674 Ok(updated)

670675 }

671676}

@@ -696,6 +701,7 @@

696701 None,

697702 )

698703 .await?;

704 self.queue_search_pull(updated.id, updated.repo_id).await?;

699705 Ok(updated)

700706 }

701707}

Modified lib/pulls/src/services/reviews.rs +3 −0 View lib/pulls/src/services/reviews.rs

@@ -153,6 +153,7 @@

153153 self.sync_reviewer(pull.id, caller, ReviewerOutcome::from_state(state))

154154 .await?;

155155 }

156 self.queue_search_pull(pull.id, pull.repo_id).await?;

156157 let _ = ctx;

157158 Ok(review)

158159 }

@@ -208,6 +209,7 @@

208209 .await?;

209210 self.sync_reviewer(pull.id, caller, ReviewerOutcome::from_state(target))

210211 .await?;

212 self.queue_search_pull(pull.id, pull.repo_id).await?;

211213 Ok(updated)

212214 }

213215

@@ -240,6 +242,7 @@

240242 .await?;

241243 self.sync_reviewer(pull.id, review.user_id, ReviewerOutcome::Dismissed)

242244 .await?;

245 self.queue_search_pull(pull.id, pull.repo_id).await?;

243246 Ok(updated)

244247 }

245248

Modified lib/pulls/src/services/service.rs +5 −0 View lib/pulls/src/services/service.rs

@@ -36,6 +36,11 @@

3636 self.state.database.write()

3737 }

3838

39 pub async fn queue_search_pull(&self, pull_id: Uuid, repo_id: Uuid) -> Result<(), PullError> {

40 search::outbox::enqueue_pull(self.db_write(), pull_id, repo_id).await?;

41 Ok(())

42 }

43

3944 pub fn uid(session: &Session) -> Result<Uuid, PullError> {

4045 use session::SessionUser;

4146 session.uid().ok_or(PullError::Unauthorized)

Modified lib/pulls/Cargo.toml +1 −0 View lib/pulls/Cargo.toml

@@ -20,3 +20,4 @@

2020time = { workspace = true }

2121tracing = { workspace = true }

2222globset = { workspace = true }

23search = { workspace = true }

Added lib/search/src/lib.rs +7 −0 View lib/search/src/lib.rs

@@ -0,0 +1,7 @@

1pub mod outbox;

2pub mod parser;

3pub mod service;

4pub mod worker;

5

6pub use service::{SearchError, SearchHit, SearchKind, SearchPage, SearchQuery, SearchService};

7pub use worker::SearchWorker;

Added lib/search/src/outbox.rs +52 −0 View lib/search/src/outbox.rs

@@ -0,0 +1,52 @@

1use models::search::index_job;

2use sea_orm::{ActiveModelTrait, ConnectionTrait, Set};

3use time::OffsetDateTime;

4use uuid::Uuid;

5

6async fn enqueue<C: ConnectionTrait>(

7 db: &C,

8 kind: &str,

9 target_id: Option<Uuid>,

10 repo_id: Option<Uuid>,

11) -> Result<(), sea_orm::DbErr> {

12 let now = OffsetDateTime::now_utc();

13 index_job::ActiveModel {

14 id: Set(Uuid::now_v7()),

15 kind: Set(kind.to_string()),

16 target_id: Set(target_id),

17 repo_id: Set(repo_id),

18 state: Set("queued".to_string()),

19 attempts: Set(0),

20 run_after: Set(now),

21 locked_at: Set(None),

22 last_error: Set(None),

23 created_at: Set(now),

24 updated_at: Set(now),

25 }

26 .insert(db)

27 .await?;

28 Ok(())

29}

30

31pub async fn enqueue_repo<C: ConnectionTrait>(db: &C, repo_id: Uuid) -> Result<(), sea_orm::DbErr> {

32 enqueue(db, "repo", Some(repo_id), Some(repo_id)).await

33}

34

35pub async fn enqueue_issue<C: ConnectionTrait>(

36 db: &C,

37 issue_id: Uuid,

38) -> Result<(), sea_orm::DbErr> {

39 enqueue(db, "issue", Some(issue_id), None).await

40}

41

42pub async fn enqueue_pull<C: ConnectionTrait>(

43 db: &C,

44 pull_id: Uuid,

45 repo_id: Uuid,

46) -> Result<(), sea_orm::DbErr> {

47 enqueue(db, "pull", Some(pull_id), Some(repo_id)).await

48}

49

50pub async fn enqueue_all<C: ConnectionTrait>(db: &C) -> Result<(), sea_orm::DbErr> {

51 enqueue(db, "all", None, None).await

52}

Added lib/search/src/parser.rs +151 −0 View lib/search/src/parser.rs

@@ -0,0 +1,151 @@

1use tree_sitter::{Language, Parser, TreeCursor};

2

3#[derive(Debug, Clone)]

4pub struct ParsedSymbol {

5 pub name: String,

6 pub kind: String,

7 pub line: u32,

8 pub end_line: u32,

9 pub is_definition: bool,

10}

11

12pub fn language_for_path(path: &str) -> Option<(&'static str, Language)> {

13 let name = path.rsplit('/').next().unwrap_or(path).to_ascii_lowercase();

14 let extension = name.rsplit_once('.').map(|(_, ext)| ext)?;

15 let language = match extension {

16 "rs" => ("Rust", tree_sitter_rust::LANGUAGE.into()),

17 "js" | "mjs" | "cjs" | "jsx" => ("JavaScript", tree_sitter_javascript::LANGUAGE.into()),

18 "ts" => (

19 "TypeScript",

20 tree_sitter_typescript::LANGUAGE_TYPESCRIPT.into(),

21 ),

22 "tsx" => ("TypeScript", tree_sitter_typescript::LANGUAGE_TSX.into()),

23 "py" => ("Python", tree_sitter_python::LANGUAGE.into()),

24 "go" => ("Go", tree_sitter_go::LANGUAGE.into()),

25 "java" => ("Java", tree_sitter_java::LANGUAGE.into()),

26 "c" | "h" => ("C", tree_sitter_c::LANGUAGE.into()),

27 "cc" | "cp" | "cpp" | "cxx" | "c++" | "hh" | "hpp" | "hxx" => {

28 ("C++", tree_sitter_cpp::LANGUAGE.into())

29 }

30 _ => return None,

31 };

32 Some(language)

33}

34

35pub fn symbols_for(path: &str, text: &str) -> Vec<ParsedSymbol> {

36 let Some((_, language)) = language_for_path(path) else {

37 return Vec::new();

38 };

39 let mut parser = Parser::new();

40 if parser.set_language(&language).is_err() {

41 return Vec::new();

42 }

43 let Some(tree) = parser.parse(text, None) else {

44 return Vec::new();

45 };

46 let bytes = text.as_bytes();

47 let mut out = Vec::new();

48 let mut cursor = tree.root_node().walk();

49 visit(&mut cursor, bytes, &mut out);

50 out

51}

52

53fn visit(cursor: &mut TreeCursor<'_>, source: &[u8], out: &mut Vec<ParsedSymbol>) {

54 let node = cursor.node();

55 let kind = node.kind();

56 if is_definition_kind(kind) {

57 if let Some(name) = declaration_name(node, source) {

58 out.push(ParsedSymbol {

59 name,

60 kind: kind.to_string(),

61 line: node.start_position().row as u32 + 1,

62 end_line: node.end_position().row as u32 + 1,

63 is_definition: true,

64 });

65 }

66 } else if is_identifier_kind(kind) {

67 if let Ok(name) = node.utf8_text(source) {

68 if !name.is_empty() {

69 out.push(ParsedSymbol {

70 name: name.to_string(),

71 kind: "reference".to_string(),

72 line: node.start_position().row as u32 + 1,

73 end_line: node.end_position().row as u32 + 1,

74 is_definition: false,

75 });

76 }

77 }

78 }

79 if cursor.goto_first_child() {

80 loop {

81 visit(cursor, source, out);

82 if !cursor.goto_next_sibling() {

83 break;

84 }

85 }

86 cursor.goto_parent();

87 }

88}

89

90fn declaration_name(node: tree_sitter::Node<'_>, source: &[u8]) -> Option<String> {

91 for field in ["name", "declarator", "type", "field"] {

92 if let Some(child) = node.child_by_field_name(field) {

93 if is_identifier_kind(child.kind()) {

94 return child.utf8_text(source).ok().map(ToOwned::to_owned);

95 }

96 if let Some(name) = first_identifier(child, source) {

97 return Some(name);

98 }

99 }

100 }

101 first_identifier(node, source)

102}

103

104fn first_identifier(node: tree_sitter::Node<'_>, source: &[u8]) -> Option<String> {

105 if is_identifier_kind(node.kind()) {

106 return node.utf8_text(source).ok().map(ToOwned::to_owned);

107 }

108 let mut cursor = node.walk();

109 if cursor.goto_first_child() {

110 loop {

111 if let Some(name) = first_identifier(cursor.node(), source) {

112 return Some(name);

113 }

114 if !cursor.goto_next_sibling() {

115 break;

116 }

117 }

118 }

119 None

120}

121

122fn is_identifier_kind(kind: &str) -> bool {

123 matches!(

124 kind,

125 "identifier" | "field_identifier" | "type_identifier" | "namespace_identifier"

126 )

127}

128

129fn is_definition_kind(kind: &str) -> bool {

130 matches!(

131 kind,

132 "function_item"

133 | "function_declaration"

134 | "function_definition"

135 | "method_definition"

136 | "method_declaration"

137 | "class_definition"

138 | "class_declaration"

139 | "interface_declaration"

140 | "struct_item"

141 | "struct_specifier"

142 | "enum_item"

143 | "enum_specifier"

144 | "trait_item"

145 | "impl_item"

146 | "type_item"

147 | "type_declaration"

148 | "module"

149 | "namespace_definition"

150 )

151}

Added lib/search/src/service.rs +1005 −0 View lib/search/src/service.rs

@@ -0,0 +1,1005 @@

1use std::collections::{HashMap, HashSet};

2use std::path::Path;

3use std::time::Duration;

4

5use config::search::SearchConfig;

6use export::AppDatabase;

7use models::dir::dirs;

8use models::issue::{comment, issue_item, issue_label, issue_relation_label, issues};

9use models::pull_request::{

10 pull_request_comment, pull_request_label, pull_request_review, pull_requests,

11};

12use models::repo::{repo_collaborations, repos};

13use models::space::{space_members, spaces};

14use models::users::user;

15use reqwest::{Client, StatusCode};

16use sea_orm::{ColumnTrait, EntityTrait, QueryFilter, QueryOrder};

17use serde::{Deserialize, Serialize};

18use serde_json::{Value, json};

19use utoipa::{IntoParams, ToSchema};

20use uuid::Uuid;

21

22use crate::parser::{language_for_path, symbols_for};

23

24#[derive(Debug, thiserror::Error)]

25pub enum SearchError {

26 #[error("search is disabled")]

27 Disabled,

28 #[error("invalid search query: {0}")]

29 Invalid(String),

30 #[error("search service request failed: {0}")]

31 Http(#[from] reqwest::Error),

32 #[error("search service returned {status}: {body}")]

33 Remote { status: StatusCode, body: String },

34 #[error("database error: {0}")]

35 Db(#[from] sea_orm::DbErr),

36 #[error("git indexing command failed: {0}")]

37 Git(String),

38 #[error("search serialization failed: {0}")]

39 Json(#[from] serde_json::Error),

40 #[error("search I/O failed: {0}")]

41 Io(#[from] std::io::Error),

42 #[error("search task timed out")]

43 TimedOut,

44}

45

46#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, ToSchema)]

47#[serde(rename_all = "snake_case")]

48pub enum SearchKind {

49 Code,

50 Symbol,

51 Issue,

52 Pull,

53}

54

55impl SearchKind {

56 fn parse_list(value: Option<&str>) -> Result<Vec<Self>, SearchError> {

57 let Some(value) = value.filter(|v| !v.trim().is_empty()) else {

58 return Ok(vec![Self::Code, Self::Symbol, Self::Issue, Self::Pull]);

59 };

60 let mut kinds = Vec::new();

61 for value in value.split(',').map(str::trim).filter(|v| !v.is_empty()) {

62 let kind = match value {

63 "code" => Self::Code,

64 "symbol" | "symbols" => Self::Symbol,

65 "issue" | "issues" => Self::Issue,

66 "pull" | "pulls" | "pr" => Self::Pull,

67 other => return Err(SearchError::Invalid(format!("unknown kind `{other}`"))),

68 };

69 if !kinds.contains(&kind) {

70 kinds.push(kind);

71 }

72 }

73 if kinds.is_empty() {

74 return Err(SearchError::Invalid("kinds must not be empty".to_string()));

75 }

76 Ok(kinds)

77 }

78

79 fn index_suffix(self) -> &'static str {

80 match self {

81 Self::Code => "code",

82 Self::Symbol => "symbols",

83 Self::Issue => "issues",

84 Self::Pull => "pulls",

85 }

86 }

87

88 fn as_str(self) -> &'static str {

89 match self {

90 Self::Code => "code",

91 Self::Symbol => "symbol",

92 Self::Issue => "issue",

93 Self::Pull => "pull",

94 }

95 }

96}

97

98#[derive(Debug, Clone, Deserialize, IntoParams)]

99pub struct SearchQuery {

100 pub q: String,

101 pub kinds: Option<String>,

102 pub space_name: Option<String>,

103 pub repo_name: Option<String>,

104 pub language: Option<String>,

105 pub state: Option<String>,

106 pub limit: Option<usize>,

107 pub offset: Option<usize>,

108}

109

110#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]

111pub struct SearchHit {

112 pub kind: SearchKind,

113 pub id: String,

114 pub title: String,

115 pub snippet: String,

116 pub space_name: Option<String>,

117 pub repo_name: Option<String>,

118 pub board_id: Option<Uuid>,

119 pub number: Option<i32>,

120 pub revision: Option<String>,

121 pub path: Option<String>,

122 pub line_start: Option<u32>,

123 pub line_end: Option<u32>,

124 pub language: Option<String>,

125 pub state: Option<String>,

126 pub symbol_id: Option<String>,

127 pub symbol_name: Option<String>,

128 pub symbol_kind: Option<String>,

129 pub definition: Option<bool>,

130}

131

132#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]

133pub struct SearchPage {

134 pub items: Vec<SearchHit>,

135 pub total: usize,

136 pub limit: usize,

137 pub offset: usize,

138}

139

140#[derive(Debug, Clone, Serialize, Deserialize, Default)]

141struct SearchDocument {

142 id: String,

143 kind: String,

144 repo_id: Option<String>,

145 space_id: Option<String>,

146 space_name: Option<String>,

147 repo_name: Option<String>,

148 board_id: Option<String>,

149 number: Option<i32>,

150 revision: Option<String>,

151 path: Option<String>,

152 line_start: Option<u32>,

153 line_end: Option<u32>,

154 language: Option<String>,

155 state: Option<String>,

156 symbol_id: Option<String>,

157 symbol_name: Option<String>,

158 symbol_kind: Option<String>,

159 definition: Option<bool>,

160 title: String,

161 content: String,

162}

163

164impl SearchDocument {

165 fn hit(self) -> SearchHit {

166 SearchHit {

167 kind: match self.kind.as_str() {

168 "symbol" => SearchKind::Symbol,

169 "issue" => SearchKind::Issue,

170 "pull" => SearchKind::Pull,

171 _ => SearchKind::Code,

172 },

173 id: self.id,

174 title: self.title,

175 snippet: compact_snippet(&self.content),

176 space_name: self.space_name,

177 repo_name: self.repo_name,

178 board_id: self.board_id.and_then(|id| Uuid::parse_str(&id).ok()),

179 number: self.number,

180 revision: self.revision,

181 path: self.path,

182 line_start: self.line_start,

183 line_end: self.line_end,

184 language: self.language,

185 state: self.state,

186 symbol_id: self.symbol_id,

187 symbol_name: self.symbol_name,

188 symbol_kind: self.symbol_kind,

189 definition: self.definition,

190 }

191 }

192}

193

194#[derive(Clone)]

195struct MeiliClient {

196 client: Client,

197 base: String,

198 api_key: String,

199 timeout: Duration,

200}

201

202impl MeiliClient {

203 fn new(config: &SearchConfig) -> Result<Self, SearchError> {

204 let client = Client::builder()

205 .timeout(Duration::from_secs(config.task_timeout_secs))

206 .build()?;

207 Ok(Self {

208 client,

209 base: config.url.trim_end_matches('/').to_string(),

210 api_key: config.api_key.clone(),

211 timeout: Duration::from_secs(config.task_timeout_secs),

212 })

213 }

214

215 fn request(&self, method: reqwest::Method, path: &str) -> reqwest::RequestBuilder {

216 self.client

217 .request(method, format!("{}{}", self.base, path))

218 .bearer_auth(&self.api_key)

219 }

220

221 async fn json(&self, request: reqwest::RequestBuilder) -> Result<Value, SearchError> {

222 let response = request.send().await?;

223 let status = response.status();

224 let body = response.text().await?;

225 if !status.is_success() {

226 return Err(SearchError::Remote { status, body });

227 }

228 Ok(serde_json::from_str(&body)?)

229 }

230

231 async fn ensure_index(&self, index: &str, searchable: &[&str]) -> Result<(), SearchError> {

232 let response = self

233 .request(reqwest::Method::POST, "/indexes")

234 .json(&json!({ "uid": index, "primaryKey": "id" }))

235 .send()

236 .await?;

237 if response.status().is_success() {

238 let task: Value = response.json().await?;

239 self.wait_task(&task).await?;

240 } else if response.status() != StatusCode::CONFLICT {

241 let status = response.status();

242 let body = response.text().await?;

243 return Err(SearchError::Remote { status, body });

244 }

245 let task = self

246 .json(

247 self.request(reqwest::Method::PATCH, &format!("/indexes/{index}/settings"))

248 .json(&json!({

249 "searchableAttributes": searchable,

250 "filterableAttributes": ["repo_id", "space_id", "space_name", "repo_name", "language", "state", "definition"],

251 "displayedAttributes": ["*"],

252 })),

253 )

254 .await?;

255 self.wait_task(&task).await

256 }

257

258 async fn add_documents(

259 &self,

260 index: &str,

261 documents: &[SearchDocument],

262 ) -> Result<(), SearchError> {

263 if documents.is_empty() {

264 return Ok(());

265 }

266 let task = self

267 .json(

268 self.request(

269 reqwest::Method::POST,

270 &format!("/indexes/{index}/documents"),

271 )

272 .json(documents),

273 )

274 .await?;

275 self.wait_task(&task).await

276 }

277

278 async fn delete_filter(&self, index: &str, filter: &str) -> Result<(), SearchError> {

279 let task = self

280 .json(

281 self.request(

282 reqwest::Method::POST,

283 &format!("/indexes/{index}/documents/delete"),

284 )

285 .json(&json!({ "filter": filter })),

286 )

287 .await?;

288 self.wait_task(&task).await

289 }

290

291 async fn delete_document(&self, index: &str, id: &str) -> Result<(), SearchError> {

292 let task = self

293 .json(self.request(

294 reqwest::Method::DELETE,

295 &format!("/indexes/{index}/documents/{id}"),

296 ))

297 .await?;

298 self.wait_task(&task).await

299 }

300

301 async fn search(

302 &self,

303 index: &str,

304 query: &str,

305 filter: Option<String>,

306 limit: usize,

307 offset: usize,

308 ) -> Result<(Vec<SearchDocument>, usize), SearchError> {

309 let value = self

310 .json(

311 self.request(reqwest::Method::POST, &format!("/indexes/{index}/search"))

312 .json(&json!({

313 "q": query,

314 "filter": filter,

315 "limit": limit,

316 "offset": offset,

317 "attributesToRetrieve": ["*"],

318 })),

319 )

320 .await?;

321 let hits = serde_json::from_value(value.get("hits").cloned().unwrap_or_else(|| json!([])))?;

322 let total = value

323 .get("estimatedTotalHits")

324 .and_then(Value::as_u64)

325 .unwrap_or_default() as usize;

326 Ok((hits, total))

327 }

328

329 async fn wait_task(&self, task: &Value) -> Result<(), SearchError> {

330 let Some(id) = task

331 .get("taskUid")

332 .or_else(|| task.get("uid"))

333 .and_then(Value::as_u64)

334 else {

335 return Ok(());

336 };

337 let deadline = std::time::Instant::now() + self.timeout;

338 loop {

339 let state = self

340 .json(self.request(reqwest::Method::GET, &format!("/tasks/{id}")))

341 .await?;

342 match state.get("status").and_then(Value::as_str) {

343 Some("succeeded") => return Ok(()),

344 Some("failed") | Some("canceled") => {

345 return Err(SearchError::Invalid(

346 state

347 .get("error")

348 .and_then(Value::as_object)

349 .and_then(|error| error.get("message"))

350 .and_then(Value::as_str)

351 .unwrap_or("Meilisearch task failed")

352 .to_string(),

353 ));

354 }

355 _ if std::time::Instant::now() >= deadline => return Err(SearchError::TimedOut),

356 _ => tokio::time::sleep(Duration::from_millis(100)).await,

357 }

358 }

359 }

360}

361

362#[derive(Clone)]

363pub struct SearchService {

364 db: std::sync::Arc<AppDatabase>,

365 config: SearchConfig,

366 meili: Option<MeiliClient>,

367}

368

369#[derive(Clone)]

370struct RepoContext {

371 repo: repos::Model,

372 space: spaces::Model,

373}

374

375impl SearchService {

376 pub fn from_parts(

377 db: std::sync::Arc<AppDatabase>,

378 config: SearchConfig,

379 ) -> Result<Self, SearchError> {

380 let meili = config

381 .enabled

382 .then(|| MeiliClient::new(&config))

383 .transpose()?;

384 Ok(Self { db, config, meili })

385 }

386

387 pub fn enabled(&self) -> bool {

388 self.meili.is_some()

389 }

390

391 pub fn config(&self) -> &SearchConfig {

392 &self.config

393 }

394

395 pub fn database(&self) -> &std::sync::Arc<AppDatabase> {

396 &self.db

397 }

398

399 fn meili(&self) -> Result<&MeiliClient, SearchError> {

400 self.meili.as_ref().ok_or(SearchError::Disabled)

401 }

402

403 fn index(&self, kind: SearchKind) -> String {

404 format!("{}_{}", self.config.index_prefix, kind.index_suffix())

405 }

406

407 async fn ensure_indexes(&self) -> Result<(), SearchError> {

408 let meili = self.meili()?;

409 meili

410 .ensure_index(

411 &self.index(SearchKind::Code),

412 &["path", "content", "language"],

413 )

414 .await?;

415 meili

416 .ensure_index(

417 &self.index(SearchKind::Symbol),

418 &["symbol_name", "path", "content"],

419 )

420 .await?;

421 meili

422 .ensure_index(&self.index(SearchKind::Issue), &["title", "content"])

423 .await?;

424 meili

425 .ensure_index(&self.index(SearchKind::Pull), &["title", "content"])

426 .await

427 }

428

429 pub async fn search(

430 &self,

431 caller: Option<Uuid>,

432 query: SearchQuery,

433 ) -> Result<SearchPage, SearchError> {

434 let q = query.q.trim();

435 if q.is_empty() {

436 return Err(SearchError::Invalid("q must not be empty".to_string()));

437 }

438 let limit = query.limit.unwrap_or(20).clamp(1, 50);

439 let offset = query.offset.unwrap_or(0);

440 let kinds = SearchKind::parse_list(query.kinds.as_deref())?;

441 let visible_repos = self.visible_repos(caller).await?;

442 let visible_repo_ids: Vec<String> = visible_repos

443 .iter()

444 .map(|ctx| ctx.repo.id.to_string())

445 .collect();

446 let visible_spaces = self.visible_space_ids(caller).await?;

447 let meili = self.meili()?;

448 let mut items = Vec::new();

449 let mut total = 0usize;

450 for kind in kinds {

451 if matches!(kind, SearchKind::Code | SearchKind::Symbol) && visible_repo_ids.is_empty()

452 {

453 continue;

454 }

455 if kind == SearchKind::Pull && visible_repo_ids.is_empty() {

456 continue;

457 }

458 if kind == SearchKind::Issue && visible_spaces.is_empty() {

459 continue;

460 }

461 let filter = build_filter(

462 kind,

463 &visible_repo_ids,

464 &visible_spaces,

465 query.space_name.as_deref(),

466 query.repo_name.as_deref(),

467 query.language.as_deref(),

468 query.state.as_deref(),

469 );

470 let (hits, count) = meili

471 .search(&self.index(kind), q, filter, limit, offset)

472 .await?;

473 total += count;

474 items.extend(hits.into_iter().map(SearchDocument::hit));

475 }

476 items.sort_by(|a, b| a.title.cmp(&b.title));

477 items.truncate(limit);

478 Ok(SearchPage {

479 items,

480 total,

481 limit,

482 offset,

483 })

484 }

485

486 pub async fn symbols(

487 &self,

488 caller: Option<Uuid>,

489 space_name: String,

490 repo_name: String,

491 q: String,

492 limit: Option<usize>,

493 offset: Option<usize>,

494 ) -> Result<SearchPage, SearchError> {

495 self.search(

496 caller,

497 SearchQuery {

498 q,

499 kinds: Some("symbol".to_string()),

500 space_name: Some(space_name),

501 repo_name: Some(repo_name),

502 language: None,

503 state: None,

504 limit,

505 offset,

506 },

507 )

508 .await

509 }

510

511 pub async fn references(

512 &self,

513 caller: Option<Uuid>,

514 space_name: String,

515 repo_name: String,

516 symbol_name: String,

517 limit: Option<usize>,

518 offset: Option<usize>,

519 ) -> Result<SearchPage, SearchError> {

520 self.symbols(caller, space_name, repo_name, symbol_name, limit, offset)

521 .await

522 }

523

524 pub async fn index_repo(&self, repo_id: Uuid, repo_root: &Path) -> Result<(), SearchError> {

525 self.ensure_indexes().await?;

526 let Some(context) = self.repo_context(repo_id).await? else {

527 return Ok(());

528 };

529 let repo_dir = repo_root.join(repo_id.to_string());

530 let sha = match git_output(

531 &repo_dir,

532 &[

533 "rev-parse",

534 &format!("refs/heads/{}", context.repo.default_branch),

535 ],

536 )

537 .await

538 {

539 Ok(value) => String::from_utf8_lossy(&value).trim().to_string(),

540 Err(_) => return Ok(()),

541 };

542 if sha.is_empty() {

543 return Ok(());

544 }

545 let meili = self.meili()?;

546 let filter = format!("repo_id = \"{}\"", repo_id);

547 meili

548 .delete_filter(&self.index(SearchKind::Code), &filter)

549 .await?;

550 meili

551 .delete_filter(&self.index(SearchKind::Symbol), &filter)

552 .await?;

553 let tree = git_output(&repo_dir, &["ls-tree", "-r", "-z", "--long", &sha]).await?;

554 let mut code_documents = Vec::new();

555 let mut symbol_documents = Vec::new();

556 for (path, size) in tree_entries(&tree) {

557 let Some((language, _)) = language_for_path(&path) else {

558 continue;

559 };

560 if size > self.config.max_file_bytes as u64 {

561 continue;

562 }

563 let content = match git_output(&repo_dir, &["show", &format!("{sha}:{path}")]).await {

564 Ok(content) => content,

565 Err(_) => continue,

566 };

567 if content.contains(&0) {

568 continue;

569 }

570 let Ok(text) = String::from_utf8(content) else {

571 continue;

572 };

573 for (start, end, chunk) in text_chunks(&text, self.config.chunk_lines) {

574 code_documents.push(SearchDocument {

575 id: Uuid::now_v7().to_string(),

576 kind: SearchKind::Code.as_str().to_string(),

577 repo_id: Some(repo_id.to_string()),

578 space_id: Some(context.space.id.to_string()),

579 space_name: Some(context.space.name.clone()),

580 repo_name: Some(context.repo.name.clone()),

581 revision: Some(sha.clone()),

582 path: Some(path.clone()),

583 line_start: Some(start),

584 line_end: Some(end),

585 language: Some(language.to_string()),

586 title: path.clone(),

587 content: chunk,

588 ..Default::default()

589 });

590 }

591 for (ordinal, symbol) in symbols_for(&path, &text).into_iter().enumerate() {

592 symbol_documents.push(SearchDocument {

593 id: Uuid::now_v7().to_string(),

594 kind: SearchKind::Symbol.as_str().to_string(),

595 repo_id: Some(repo_id.to_string()),

596 space_id: Some(context.space.id.to_string()),

597 space_name: Some(context.space.name.clone()),

598 repo_name: Some(context.repo.name.clone()),

599 revision: Some(sha.clone()),

600 path: Some(path.clone()),

601 line_start: Some(symbol.line),

602 line_end: Some(symbol.end_line),

603 language: Some(language.to_string()),

604 symbol_id: Some(format!("{repo_id}:{sha}:{path}:{}:{ordinal}", symbol.line)),

605 symbol_name: Some(symbol.name.clone()),

606 symbol_kind: Some(symbol.kind.clone()),

607 definition: Some(symbol.is_definition),

608 title: symbol.name,

609 content: format!("{}:{}", path, symbol.line),

610 ..Default::default()

611 });

612 }

613 }

614 for batch in code_documents.chunks(self.config.batch_size) {

615 meili

616 .add_documents(&self.index(SearchKind::Code), batch)

617 .await?;

618 }

619 for batch in symbol_documents.chunks(self.config.batch_size) {

620 meili

621 .add_documents(&self.index(SearchKind::Symbol), batch)

622 .await?;

623 }

624 Ok(())

625 }

626

627 pub async fn index_issue(&self, issue_id: Uuid) -> Result<(), SearchError> {

628 self.ensure_indexes().await?;

629 let meili = self.meili()?;

630 let Some(issue) = issue_item::Entity::find_by_id(issue_id)

631 .one(self.db.read())

632 .await?

633 else {

634 return meili

635 .delete_document(&self.index(SearchKind::Issue), &issue_id.to_string())

636 .await;

637 };

638 let Some(board) = issues::Entity::find_by_id(issue.collection_id)

639 .one(self.db.read())

640 .await?

641 else {

642 return Ok(());

643 };

644 let Some(space) = self.space_for_dir(board.dir_id).await? else {

645 return Ok(());

646 };

647 let comments = comment::Entity::find()

648 .filter(comment::Column::IssueId.eq(issue_id))

649 .order_by_asc(comment::Column::CreatedAt)

650 .all(self.db.read())

651 .await?;

652 let label_ids: Vec<Uuid> = issue_relation_label::Entity::find()

653 .filter(issue_relation_label::Column::IssueId.eq(issue_id))

654 .all(self.db.read())

655 .await?

656 .into_iter()

657 .map(|row| row.label_id)

658 .collect();

659 let labels = if label_ids.is_empty() {

660 Vec::new()

661 } else {

662 issue_label::Entity::find()

663 .filter(issue_label::Column::Id.is_in(label_ids))

664 .all(self.db.read())

665 .await?

666 .into_iter()

667 .map(|row| row.name)

668 .collect()

669 };

670 let author = user::Entity::find_by_id(issue.author_id)

671 .one(self.db.read())

672 .await?;

673 let mut content = vec![issue.description.unwrap_or_default(), issue.body];

674 content.extend(labels);

675 content.extend(comments.into_iter().map(|row| row.content));

676 let document = SearchDocument {

677 id: issue.id.to_string(),

678 kind: SearchKind::Issue.as_str().to_string(),

679 space_id: Some(space.id.to_string()),

680 space_name: Some(space.name),

681 board_id: Some(board.id.to_string()),

682 number: Some(issue.autoincrement_id),

683 state: Some(

684 if issue.closed_at.is_some() {

685 "closed"

686 } else {

687 "open"

688 }

689 .to_string(),

690 ),

691 title: issue.title,

692 content: format!(

693 "{}\n{}",

694 author.map(|row| row.username).unwrap_or_default(),

695 content.join("\n")

696 ),

697 ..Default::default()

698 };

699 meili

700 .add_documents(&self.index(SearchKind::Issue), &[document])

701 .await

702 }

703

704 pub async fn index_pull(&self, pull_id: Uuid) -> Result<(), SearchError> {

705 self.ensure_indexes().await?;

706 let meili = self.meili()?;

707 let Some(pull) = pull_requests::Entity::find_by_id(pull_id)

708 .one(self.db.read())

709 .await?

710 else {

711 return meili

712 .delete_document(&self.index(SearchKind::Pull), &pull_id.to_string())

713 .await;

714 };

715 let Some(context) = self.repo_context(pull.repo_id).await? else {

716 return Ok(());

717 };

718 let comments = pull_request_comment::Entity::find()

719 .filter(pull_request_comment::Column::PullRequestId.eq(pull.id))

720 .order_by_asc(pull_request_comment::Column::CreatedAt)

721 .all(self.db.read())

722 .await?;

723 let reviews = pull_request_review::Entity::find()

724 .filter(pull_request_review::Column::PullRequestId.eq(pull.id))

725 .order_by_asc(pull_request_review::Column::CreatedAt)

726 .all(self.db.read())

727 .await?;

728 let label_ids: Vec<Uuid> = pull_request_label::Entity::find()

729 .filter(pull_request_label::Column::PullRequestId.eq(pull.id))

730 .all(self.db.read())

731 .await?

732 .into_iter()

733 .map(|row| row.label_id)

734 .collect();

735 let labels = if label_ids.is_empty() {

736 Vec::new()

737 } else {

738 issue_label::Entity::find()

739 .filter(issue_label::Column::Id.is_in(label_ids))

740 .all(self.db.read())

741 .await?

742 .into_iter()

743 .map(|row| row.name)

744 .collect()

745 };

746 let author = user::Entity::find_by_id(pull.author_id)

747 .one(self.db.read())

748 .await?;

749 let mut content = vec![pull.body];

750 content.extend(labels);

751 content.extend(comments.into_iter().map(|row| row.body));

752 content.extend(reviews.into_iter().filter_map(|row| row.content));

753 let state = if pull.has_merged {

754 "merged"

755 } else if pull.closed_at.is_some() {

756 "closed"

757 } else {

758 "open"

759 };

760 let document = SearchDocument {

761 id: pull.id.to_string(),

762 kind: SearchKind::Pull.as_str().to_string(),

763 repo_id: Some(context.repo.id.to_string()),

764 space_id: Some(context.space.id.to_string()),

765 space_name: Some(context.space.name),

766 repo_name: Some(context.repo.name),

767 number: Some(pull.autoincrement_id),

768 state: Some(state.to_string()),

769 title: pull.title,

770 content: format!(

771 "{}\n{}",

772 author.map(|row| row.username).unwrap_or_default(),

773 content.join("\n")

774 ),

775 ..Default::default()

776 };

777 meili

778 .add_documents(&self.index(SearchKind::Pull), &[document])

779 .await

780 }

781

782 pub async fn queue_all(&self) -> Result<(), SearchError> {

783 crate::outbox::enqueue_all(self.db.write()).await?;

784 Ok(())

785 }

786

787 pub async fn queue_initial(&self) -> Result<(), SearchError> {

788 if !self.enabled() {

789 return Ok(());

790 }

791 let has_work = models::search::index_job::Entity::find()

792 .filter(models::search::index_job::Column::State.eq("queued"))

793 .one(self.db.read())

794 .await?

795 .is_some();

796 if !has_work {

797 self.queue_all().await?;

798 }

799 Ok(())

800 }

801

802 async fn repo_context(&self, repo_id: Uuid) -> Result<Option<RepoContext>, SearchError> {

803 let Some(repo) = repos::Entity::find_by_id(repo_id)

804 .one(self.db.read())

805 .await?

806 else {

807 return Ok(None);

808 };

809 let Some(space) = self.space_for_dir(repo.dir_id).await? else {

810 return Ok(None);

811 };

812 Ok(Some(RepoContext { repo, space }))

813 }

814

815 async fn space_for_dir(&self, dir_id: Uuid) -> Result<Option<spaces::Model>, SearchError> {

816 let Some(dir) = dirs::Entity::find_by_id(dir_id).one(self.db.read()).await? else {

817 return Ok(None);

818 };

819 let Some(space_id) = dir.space_id else {

820 return Ok(None);

821 };

822 Ok(spaces::Entity::find_by_id(space_id)

823 .one(self.db.read())

824 .await?)

825 }

826

827 async fn visible_repos(&self, caller: Option<Uuid>) -> Result<Vec<RepoContext>, SearchError> {

828 let repos = repos::Entity::find()

829 .filter(repos::Column::IsArchived.eq(false))

830 .all(self.db.read())

831 .await?;

832 let dir_ids: Vec<Uuid> = repos.iter().map(|repo| repo.dir_id).collect();

833 let dirs = dirs::Entity::find()

834 .filter(dirs::Column::Id.is_in(dir_ids))

835 .all(self.db.read())

836 .await?;

837 let space_ids: Vec<Uuid> = dirs.iter().filter_map(|dir| dir.space_id).collect();

838 let spaces = spaces::Entity::find()

839 .filter(spaces::Column::Id.is_in(space_ids))

840 .all(self.db.read())

841 .await?;

842 let dirs: HashMap<Uuid, dirs::Model> = dirs.into_iter().map(|dir| (dir.id, dir)).collect();

843 let spaces: HashMap<Uuid, spaces::Model> =

844 spaces.into_iter().map(|space| (space.id, space)).collect();

845 let collaborations: HashSet<Uuid> = match caller {

846 Some(uid) => repo_collaborations::Entity::find()

847 .filter(repo_collaborations::Column::UserId.eq(uid))

848 .all(self.db.read())

849 .await?

850 .into_iter()

851 .map(|row| row.repo_id)

852 .collect(),

853 None => HashSet::new(),

854 };

855 Ok(repos

856 .into_iter()

857 .filter_map(|repo| {

858 let space = dirs

859 .get(&repo.dir_id)?

860 .space_id

861 .and_then(|id| spaces.get(&id))?

862 .clone();

863 let visible = !repo.is_private

864 || caller.is_some_and(|uid| uid == space.owner_id)

865 || collaborations.contains(&repo.id);

866 visible.then_some(RepoContext { repo, space })

867 })

868 .collect())

869 }

870

871 async fn visible_space_ids(&self, caller: Option<Uuid>) -> Result<Vec<String>, SearchError> {

872 let Some(caller) = caller else {

873 return Ok(Vec::new());

874 };

875 let owned = spaces::Entity::find()

876 .filter(spaces::Column::OwnerId.eq(caller))

877 .all(self.db.read())

878 .await?;

879 let member_ids: Vec<Uuid> = space_members::Entity::find()

880 .filter(space_members::Column::UserId.eq(caller))

881 .all(self.db.read())

882 .await?

883 .into_iter()

884 .map(|row| row.space_id)

885 .collect();

886 let mut ids: HashSet<String> = owned

887 .into_iter()

888 .map(|space| space.id.to_string())

889 .collect();

890 ids.extend(member_ids.into_iter().map(|id| id.to_string()));

891 Ok(ids.into_iter().collect())

892 }

893}

894

895fn build_filter(

896 kind: SearchKind,

897 repo_ids: &[String],

898 space_ids: &[String],

899 space_name: Option<&str>,

900 repo_name: Option<&str>,

901 language: Option<&str>,

902 state: Option<&str>,

903) -> Option<String> {

904 let mut filters = Vec::new();

905 if matches!(

906 kind,

907 SearchKind::Code | SearchKind::Symbol | SearchKind::Pull

908 ) {

909 filters.push(in_filter("repo_id", repo_ids));

910 } else {

911 filters.push(in_filter("space_id", space_ids));

912 }

913 if let Some(value) = space_name.filter(|value| !value.trim().is_empty()) {

914 filters.push(eq_filter("space_name", value));

915 }

916 if let Some(value) = repo_name.filter(|value| !value.trim().is_empty()) {

917 filters.push(eq_filter("repo_name", value));

918 }

919 if let Some(value) = language.filter(|value| !value.trim().is_empty()) {

920 filters.push(eq_filter("language", value));

921 }

922 if let Some(value) = state.filter(|value| !value.trim().is_empty()) {

923 filters.push(eq_filter("state", value));

924 }

925 Some(filters.join(" AND "))

926}

927

928fn in_filter(field: &str, values: &[String]) -> String {

929 let values = values

930 .iter()

931 .map(|value| format!("\"{}\"", escape_filter(value)))

932 .collect::<Vec<_>>()

933 .join(", ");

934 format!("{field} IN [{values}]")

935}

936

937fn eq_filter(field: &str, value: &str) -> String {

938 format!("{field} = \"{}\"", escape_filter(value))

939}

940

941fn escape_filter(value: &str) -> String {

942 value.replace('\\', "\\\\").replace('"', "\\\"")

943}

944

945async fn git_output(repo_dir: &Path, args: &[&str]) -> Result<Vec<u8>, SearchError> {

946 let output = tokio::process::Command::new("git")

947 .arg("--git-dir")

948 .arg(repo_dir)

949 .args(args)

950 .output()

951 .await?;

952 if output.status.success() {

953 Ok(output.stdout)

954 } else {

955 Err(SearchError::Git(

956 String::from_utf8_lossy(&output.stderr).trim().to_string(),

957 ))

958 }

959}

960

961fn tree_entries(value: &[u8]) -> Vec<(String, u64)> {

962 value

963 .split(|byte| *byte == 0)

964 .filter_map(|entry| {

965 let entry = std::str::from_utf8(entry).ok()?;

966 let (meta, path) = entry.split_once('\t')?;

967 let mut fields = meta.split_ascii_whitespace();

968 let _mode = fields.next()?;

969 let kind = fields.next()?;

970 let _oid = fields.next()?;

971 let size = fields.next()?.parse().ok()?;

972 (kind == "blob").then_some((path.to_string(), size))

973 })

974 .collect()

975}

976

977fn text_chunks(text: &str, chunk_lines: usize) -> Vec<(u32, u32, String)> {

978 let lines: Vec<&str> = text.lines().collect();

979 if lines.is_empty() {

980 return Vec::new();

981 }

982 lines

983 .chunks(chunk_lines)

984 .enumerate()

985 .map(|(index, chunk)| {

986 let start = index * chunk_lines + 1;

987 (

988 start as u32,

989 (start + chunk.len() - 1) as u32,

990 chunk.join("\n"),

991 )

992 })

993 .collect()

994}

995

996fn compact_snippet(value: &str) -> String {

997 let normalized = value.split_whitespace().collect::<Vec<_>>().join(" ");

998 let mut chars = normalized.chars();

999 let snippet: String = chars.by_ref().take(280).collect();

1000 if chars.next().is_some() {

1001 format!("{snippet}…")

1002 } else {

1003 snippet

1004 }

1005}

Added lib/search/src/worker.rs +187 −0 View lib/search/src/worker.rs

@@ -0,0 +1,187 @@

1use std::path::PathBuf;

2use std::sync::Arc;

3use std::time::Duration;

4

5use export::AppDatabase;

6use models::issue::issue_item;

7use models::pull_request::pull_requests;

8use models::repo::repos;

9use models::search::index_job;

10use sea_orm::{ColumnTrait, EntityTrait, QueryFilter, QueryOrder};

11use time::OffsetDateTime;

12use tokio_util::sync::CancellationToken;

13use tracing::{error, info, warn};

14

15use crate::outbox;

16use crate::{SearchError, SearchService};

17

18#[derive(Clone)]

19pub struct SearchWorker {

20 db: Arc<AppDatabase>,

21 service: SearchService,

22 repo_root: PathBuf,

23 cancel: CancellationToken,

24}

25

26impl SearchWorker {

27 pub fn new(

28 db: Arc<AppDatabase>,

29 service: SearchService,

30 repo_root: PathBuf,

31 cancel: CancellationToken,

32 ) -> Self {

33 Self {

34 db,

35 service,

36 repo_root,

37 cancel,

38 }

39 }

40

41 pub async fn run(self) {

42 if let Err(error) = self.service.queue_initial().await {

43 warn!(error = %error, "search initial queue failed");

44 }

45 let delay = Duration::from_secs(self.service.config().poll_interval_secs);

46 loop {

47 tokio::select! {

48 _ = self.cancel.cancelled() => break,

49 _ = tokio::time::sleep(delay) => match self.run_once().await {

50 Ok(true) => {}

51 Ok(false) => {}

52 Err(error) => warn!(error = %error, "search worker iteration failed"),

53 }

54 }

55 }

56 info!("search worker stopped");

57 }

58

59 pub async fn run_once(&self) -> Result<bool, SearchError> {

60 if !self.service.enabled() {

61 return Ok(false);

62 }

63 let now = OffsetDateTime::now_utc();

64 let Some(job) = index_job::Entity::find()

65 .filter(index_job::Column::State.eq("queued"))

66 .filter(index_job::Column::RunAfter.lte(now))

67 .order_by_asc(index_job::Column::CreatedAt)

68 .one(self.db.read())

69 .await?

70 else {

71 return Ok(false);

72 };

73 let claimed = index_job::Entity::update_many()

74 .col_expr(

75 index_job::Column::State,

76 sea_orm::sea_query::Expr::value("processing"),

77 )

78 .col_expr(

79 index_job::Column::LockedAt,

80 sea_orm::sea_query::Expr::value(now),

81 )

82 .col_expr(

83 index_job::Column::UpdatedAt,

84 sea_orm::sea_query::Expr::value(now),

85 )

86 .filter(index_job::Column::Id.eq(job.id))

87 .filter(index_job::Column::State.eq("queued"))

88 .exec(self.db.write())

89 .await?;

90 if claimed.rows_affected == 0 {

91 return Ok(false);

92 }

93 let result = self.process(&job).await;

94 match result {

95 Ok(()) => {

96 index_job::Entity::update_many()

97 .col_expr(

98 index_job::Column::State,

99 sea_orm::sea_query::Expr::value("succeeded"),

100 )

101 .col_expr(

102 index_job::Column::UpdatedAt,

103 sea_orm::sea_query::Expr::value(OffsetDateTime::now_utc()),

104 )

105 .filter(index_job::Column::Id.eq(job.id))

106 .exec(self.db.write())

107 .await?;

108 info!(job_id = %job.id, kind = %job.kind, "search index job completed");

109 }

110 Err(error) => {

111 let attempts = job.attempts + 1;

112 let retry_after = OffsetDateTime::now_utc()

113 + time::Duration::seconds((attempts.min(8) * 5) as i64);

114 index_job::Entity::update_many()

115 .col_expr(

116 index_job::Column::State,

117 sea_orm::sea_query::Expr::value("queued"),

118 )

119 .col_expr(

120 index_job::Column::Attempts,

121 sea_orm::sea_query::Expr::value(attempts),

122 )

123 .col_expr(

124 index_job::Column::RunAfter,

125 sea_orm::sea_query::Expr::value(retry_after),

126 )

127 .col_expr(

128 index_job::Column::LockedAt,

129 sea_orm::sea_query::Expr::value(Option::<OffsetDateTime>::None),

130 )

131 .col_expr(

132 index_job::Column::LastError,

133 sea_orm::sea_query::Expr::value(error.to_string()),

134 )

135 .col_expr(

136 index_job::Column::UpdatedAt,

137 sea_orm::sea_query::Expr::value(OffsetDateTime::now_utc()),

138 )

139 .filter(index_job::Column::Id.eq(job.id))

140 .exec(self.db.write())

141 .await?;

142 error!(job_id = %job.id, kind = %job.kind, error = %error, "search index job failed and was requeued");

143 }

144 }

145 Ok(true)

146 }

147

148 async fn process(&self, job: &index_job::Model) -> Result<(), SearchError> {

149 match job.kind.as_str() {

150 "repo" => {

151 let repo_id = job

152 .target_id

153 .ok_or_else(|| SearchError::Invalid("repo job has no target".to_string()))?;

154 self.service.index_repo(repo_id, &self.repo_root).await

155 }

156 "issue" => {

157 let issue_id = job

158 .target_id

159 .ok_or_else(|| SearchError::Invalid("issue job has no target".to_string()))?;

160 self.service.index_issue(issue_id).await

161 }

162 "pull" => {

163 let pull_id = job

164 .target_id

165 .ok_or_else(|| SearchError::Invalid("pull job has no target".to_string()))?;

166 self.service.index_pull(pull_id).await

167 }

168 "all" => self.expand_all().await,

169 other => Err(SearchError::Invalid(format!(

170 "unknown index job kind `{other}`"

171 ))),

172 }

173 }

174

175 async fn expand_all(&self) -> Result<(), SearchError> {

176 for repo in repos::Entity::find().all(self.db.read()).await? {

177 outbox::enqueue_repo(self.db.write(), repo.id).await?;

178 }

179 for issue in issue_item::Entity::find().all(self.db.read()).await? {

180 outbox::enqueue_issue(self.db.write(), issue.id).await?;

181 }

182 for pull in pull_requests::Entity::find().all(self.db.read()).await? {

183 outbox::enqueue_pull(self.db.write(), pull.id, pull.repo_id).await?;

184 }

185 Ok(())

186 }

187}

Added lib/search/Cargo.toml +29 −0 View lib/search/Cargo.toml

@@ -0,0 +1,29 @@

1[package]

2name = "search"

3version.workspace = true

4edition.workspace = true

5

6[dependencies]

7config = { workspace = true }

8export = { workspace = true }

9models = { workspace = true }

10reqwest = { workspace = true }

11sea-orm = { workspace = true }

12serde = { workspace = true, features = ["derive"] }

13serde_json = { workspace = true }

14thiserror = { workspace = true }

15time = { workspace = true }

16tokio = { workspace = true, features = ["process", "time", "sync", "rt", "macros"] }

17tokio-util = { workspace = true }

18tracing = { workspace = true }

19tree-sitter = { workspace = true }

20tree-sitter-rust = { workspace = true }

21tree-sitter-javascript = { workspace = true }

22tree-sitter-typescript = { workspace = true }

23tree-sitter-python = { workspace = true }

24tree-sitter-go = { workspace = true }

25tree-sitter-java = { workspace = true }

26tree-sitter-c = { workspace = true }

27tree-sitter-cpp = { workspace = true }

28utoipa = { workspace = true }

29uuid = { workspace = true }

Modified lib/webapi/src/lib.rs +7 −1 View lib/webapi/src/lib.rs

@@ -1,5 +1,6 @@

11pub mod metrics;

22pub mod openapi;

3pub mod search;

34pub mod spa;

45

56use axum::Router;

@@ -54,6 +55,9 @@

5455 let user = user::services::service::UserService::new(state.clone());

5556 let git = git::service::GitService::new(state.clone());

5657 let pulls = pulls::services::service::PullService::new(state.clone());

58 let search =

59 ::search::SearchService::from_parts(state.database.clone(), state.config.search.clone())

60 .expect("validated search client configuration");

5761 let health_state = state;

5862

5963 let public: Router = Router::new()

@@ -63,9 +67,11 @@

6367 .merge(space::routes::router(space.clone()).with_state(space))

6468 .merge(user::routes::router(user.clone()).with_state(user))

6569 .merge(git::routes::router(git.clone()).with_state(git.clone()))

66 .merge(pulls::routes::router().with_state(pulls));

70 .merge(pulls::routes::router().with_state(pulls))

71 .merge(crate::search::router().with_state(search.clone()));

6772 let internal: Router = Router::new()

6873 .merge(git::routes::internal_router(git.clone()).with_state(git))

74 .merge(crate::search::internal_router().with_state(search))

6975 .route_layer(axum::middleware::from_fn_with_state(

7076 auth.clone(),

7177 auth::share::guard::admin_guard,

Modified lib/webapi/src/openapi.rs +1 −0 View lib/webapi/src/openapi.rs

@@ -111,6 +111,7 @@

111111 m!("user", user::openapi::UserApiDoc::openapi());

112112 m!("git", git::openapi::GitApiDoc::openapi());

113113 m!("pulls", pulls::openapi::PullsApiDoc::openapi());

114 m!("search", crate::search::SearchApiDoc::openapi());

114115 doc.servers = Some(vec![utoipa::openapi::Server::new("/api/v1")]);

115116 doc.info.description = Some(CONVENTIONS.to_string());

116117 doc.info.license = None;

Added lib/webapi/src/search.rs +181 −0 View lib/webapi/src/search.rs

@@ -0,0 +1,181 @@

1use axum::extract::{Path, Query, State};

2use axum::http::StatusCode;

3use axum::response::{IntoResponse, Response};

4use axum::routing::{get, post};

5use axum::{Json, Router};

6use export::rest::{AppResponseData, NoData};

7use serde::Deserialize;

8use session::{Session, SessionUser};

9use utoipa::{IntoParams, OpenApi, ToSchema};

10use uuid::Uuid;

11

12use ::search::{SearchError, SearchPage, SearchQuery, SearchService};

13

14#[derive(Debug, Clone, Deserialize, IntoParams)]

15pub struct SymbolQuery {

16 pub q: String,

17 pub limit: Option<usize>,

18 pub offset: Option<usize>,

19}

20

21#[derive(Debug, Clone, Deserialize, ToSchema)]

22pub struct ReindexCommand {

23 pub repo_id: Option<Uuid>,

24}

25

26#[derive(Debug)]

27pub struct SearchApiError(SearchError);

28

29impl From<SearchError> for SearchApiError {

30 fn from(value: SearchError) -> Self {

31 Self(value)

32 }

33}

34

35impl IntoResponse for SearchApiError {

36 fn into_response(self) -> Response {

37 let (status, message, code) = match &self.0 {

38 SearchError::Disabled => (

39 StatusCode::SERVICE_UNAVAILABLE,

40 "search is unavailable",

41 "search_disabled",

42 ),

43 SearchError::Invalid(_) => (

44 StatusCode::BAD_REQUEST,

45 "invalid search request",

46 "invalid_search_query",

47 ),

48 SearchError::Remote { .. } | SearchError::TimedOut => (

49 StatusCode::SERVICE_UNAVAILABLE,

50 "search is temporarily unavailable",

51 "search_unavailable",

52 ),

53 _ => (

54 StatusCode::INTERNAL_SERVER_ERROR,

55 "search request failed",

56 "search_failed",

57 ),

58 };

59 (

60 status,

61 AppResponseData::<NoData>::err(status.as_u16() as i32, message)

62 .with_error(self.0.to_string())

63 .with_error_code(code),

64 )

65 .into_response()

66 }

67}

68

69#[utoipa::path(

70 get,

71 path = "/search",

72 tag = "search",

73 description = "全局检索当前调用者可见的代码、符号、议题和合并请求。",

74 params(SearchQuery),

75 responses((status = 200, description = "search results", body = AppResponseData<SearchPage>))

76)]

77pub async fn global_search(

78 State(service): State<SearchService>,

79 session: Session,

80 Query(query): Query<SearchQuery>,

81) -> Result<AppResponseData<SearchPage>, SearchApiError> {

82 Ok(AppResponseData::ok(

83 service.search(session.uid(), query).await?,

84 ))

85}

86

87#[utoipa::path(

88 get,

89 path = "/repos/{space_name}/{repo_name}/symbols",

90 tag = "search",

91 description = "检索仓库默认分支中已索引的符号。",

92 params(git::handler::RepoPath, SymbolQuery),

93 responses((status = 200, description = "symbol results", body = AppResponseData<SearchPage>))

94)]

95pub async fn repo_symbols(

96 State(service): State<SearchService>,

97 session: Session,

98 Path(path): Path<git::handler::RepoPath>,

99 Query(query): Query<SymbolQuery>,

100) -> Result<AppResponseData<SearchPage>, SearchApiError> {

101 Ok(AppResponseData::ok(

102 service

103 .symbols(

104 session.uid(),

105 path.space_name,

106 path.repo_name,

107 query.q,

108 query.limit,

109 query.offset,

110 )

111 .await?,

112 ))

113}

114

115#[utoipa::path(

116 get,

117 path = "/repos/{space_name}/{repo_name}/symbols/{symbol_name}/references",

118 tag = "search",

119 description = "返回仓库内同名符号的词法引用与定义候选。",

120 params(git::handler::RepoPath, SymbolQuery),

121 responses((status = 200, description = "symbol references", body = AppResponseData<SearchPage>))

122)]

123pub async fn symbol_references(

124 State(service): State<SearchService>,

125 session: Session,

126 Path((space_name, repo_name, symbol_name)): Path<(String, String, String)>,

127 Query(query): Query<SymbolQuery>,

128) -> Result<AppResponseData<SearchPage>, SearchApiError> {

129 let name = if query.q.trim().is_empty() {

130 symbol_name

131 } else {

132 query.q

133 };

134 Ok(AppResponseData::ok(

135 service

136 .references(

137 session.uid(),

138 space_name,

139 repo_name,

140 name,

141 query.limit,

142 query.offset,

143 )

144 .await?,

145 ))

146}

147

148pub async fn reindex(

149 State(service): State<SearchService>,

150 Json(command): Json<ReindexCommand>,

151) -> Result<AppResponseData<NoData>, SearchApiError> {

152 if let Some(repo_id) = command.repo_id {

153 ::search::outbox::enqueue_repo(service.database().write(), repo_id)

154 .await

155 .map_err(SearchError::from)?;

156 } else {

157 service.queue_all().await?;

158 }

159 Ok(AppResponseData::empty())

160}

161

162pub fn router() -> Router<SearchService> {

163 Router::new()

164 .route("/search", get(global_search))

165 .route("/repos/{space_name}/{repo_name}/symbols", get(repo_symbols))

166 .route(

167 "/repos/{space_name}/{repo_name}/symbols/{symbol_name}/references",

168 get(symbol_references),

169 )

170}

171

172pub fn internal_router() -> Router<SearchService> {

173 Router::new().route("/search/reindex", post(reindex))

174}

175

176#[derive(OpenApi)]

177#[openapi(

178 paths(global_search, repo_symbols, symbol_references),

179 components(schemas(SearchPage, ::search::SearchHit, ::search::SearchKind))

180)]

181pub struct SearchApiDoc;

Modified lib/webapi/Cargo.toml +4 −1 View lib/webapi/Cargo.toml

@@ -12,6 +12,7 @@

1212space = { workspace = true }

1313user = { workspace = true }

1414git = { workspace = true }

15search = { workspace = true }

1516export = { workspace = true }

1617session = { workspace = true }

1718axum = { workspace = true }

@@ -23,4 +24,6 @@

2324opentelemetry = { workspace = true }

2425opentelemetry_sdk = { workspace = true }

2526opentelemetry-otlp = { workspace = true }

26web = { git = "https://gitea.gitdata.me/GitDataAi/web.git", rev = "cd8cdac92ffbf482b5318f93848f82ed4140b7a6" }

No newline at end of file

27web = { git = "https://gitea.gitdata.me/GitDataAi/web.git", rev = "cd8cdac92ffbf482b5318f93848f82ed4140b7a6" }

28serde = { workspace = true, features = ["derive"] }

29uuid = { workspace = true }

Modified Cargo.lock +143 −0 View Cargo.lock

@@ -2627,6 +2627,7 @@

26272627 "redis",

26282628 "russh",

26292629 "sea-orm",

2630 "search",

26302631 "serde",

26312632 "serde_json",

26322633 "session",

@@ -4628,6 +4629,7 @@

46284629 "export",

46294630 "models",

46304631 "sea-orm",

4632 "search",

46314633 "serde",

46324634 "serde_json",

46334635 "session",

@@ -6299,6 +6301,7 @@

62996301 "issues",

63006302 "models",

63016303 "sea-orm",

6304 "search",

63026305 "serde",

63036306 "serde_json",

63046307 "session",

@@ -7294,6 +7297,35 @@

72947297checksum = "1c107b6f4780854c8b126e228ea8869f4d7b71260f962fefb57b996b8959ba6b"

72957298

72967299[[package]]

7300name = "search"

7301version = "1.0.0"

7302dependencies = [

7303 "config",

7304 "export",

7305 "models",

7306 "reqwest 0.12.28",

7307 "sea-orm",

7308 "serde",

7309 "serde_json",

7310 "thiserror 2.0.20",

7311 "time 0.3.55",

7312 "tokio",

7313 "tokio-util",

7314 "tracing",

7315 "tree-sitter",

7316 "tree-sitter-c",

7317 "tree-sitter-cpp",

7318 "tree-sitter-go",

7319 "tree-sitter-java",

7320 "tree-sitter-javascript",

7321 "tree-sitter-python",

7322 "tree-sitter-rust",

7323 "tree-sitter-typescript",

7324 "utoipa",

7325 "uuid",

7326]

7327

7328[[package]]

72977329name = "sec1"

72987330version = "0.7.3"

72997331source = "registry+https://github.com/rust-lang/crates.io-index"

@@ -7386,6 +7418,7 @@

73867418source = "registry+https://github.com/rust-lang/crates.io-index"

73877419checksum = "c841b55ecdae098c80dcae9cf767f6f8a0c2cdb3416bbef72181df4d0fe73f14"

73887420dependencies = [

7421 "indexmap",

73897422 "itoa",

73907423 "memchr",

73917424 "serde",

@@ -8074,6 +8107,12 @@

80748107]

80758108

80768109[[package]]

8110name = "streaming-iterator"

8111version = "0.1.9"

8112source = "registry+https://github.com/rust-lang/crates.io-index"

8113checksum = "2b2231b7c3057d5e4ad0156fb3dc807d900806020c5ffa3ee6ff2c8c76fb8520"

8114

8115[[package]]

80778116name = "stringprep"

80788117version = "0.1.5"

80798118source = "registry+https://github.com/rust-lang/crates.io-index"

@@ -8188,6 +8227,7 @@

81888227 "config",

81898228 "export",

81908229 "git",

8230 "search",

81918231 "tokio",

81928232 "tokio-util",

81938233 "tracing",

@@ -8644,6 +8684,106 @@

86448684checksum = "04a79e25382e2e852e8da874249358d382ebaf259d0d34e75d8db16a7efabbc7"

86458685

86468686[[package]]

8687name = "tree-sitter"

8688version = "0.25.10"

8689source = "registry+https://github.com/rust-lang/crates.io-index"

8690checksum = "78f873475d258561b06f1c595d93308a7ed124d9977cb26b148c2084a4a3cc87"

8691dependencies = [

8692 "cc",

8693 "regex",

8694 "regex-syntax",

8695 "serde_json",

8696 "streaming-iterator",

8697 "tree-sitter-language",

8698]

8699

8700[[package]]

8701name = "tree-sitter-c"

8702version = "0.24.2"

8703source = "registry+https://github.com/rust-lang/crates.io-index"

8704checksum = "a9b2eb57a55fed6b00812912e730b7a275cf4fe98bfd6a5d76263d4438371728"

8705dependencies = [

8706 "cc",

8707 "tree-sitter-language",

8708]

8709

8710[[package]]

8711name = "tree-sitter-cpp"

8712version = "0.23.4"

8713source = "registry+https://github.com/rust-lang/crates.io-index"

8714checksum = "df2196ea9d47b4ab4a31b9297eaa5a5d19a0b121dceb9f118f6790ad0ab94743"

8715dependencies = [

8716 "cc",

8717 "tree-sitter-language",

8718]

8719

8720[[package]]

8721name = "tree-sitter-go"

8722version = "0.25.0"

8723source = "registry+https://github.com/rust-lang/crates.io-index"

8724checksum = "c8560a4d2f835cc0d4d2c2e03cbd0dde2f6114b43bc491164238d333e28b16ea"

8725dependencies = [

8726 "cc",

8727 "tree-sitter-language",

8728]

8729

8730[[package]]

8731name = "tree-sitter-java"

8732version = "0.23.5"

8733source = "registry+https://github.com/rust-lang/crates.io-index"

8734checksum = "0aa6cbcdc8c679b214e616fd3300da67da0e492e066df01bcf5a5921a71e90d6"

8735dependencies = [

8736 "cc",

8737 "tree-sitter-language",

8738]

8739

8740[[package]]

8741name = "tree-sitter-javascript"

8742version = "0.25.0"

8743source = "registry+https://github.com/rust-lang/crates.io-index"

8744checksum = "68204f2abc0627a90bdf06e605f5c470aa26fdcb2081ea553a04bdad756693f5"

8745dependencies = [

8746 "cc",

8747 "tree-sitter-language",

8748]

8749

8750[[package]]

8751name = "tree-sitter-language"

8752version = "0.1.8"

8753source = "registry+https://github.com/rust-lang/crates.io-index"

8754checksum = "ca0d1bf6fdd806e43ae5198f82f527056d359def39e54e67a0f478ac09dac081"

8755

8756[[package]]

8757name = "tree-sitter-python"

8758version = "0.25.0"

8759source = "registry+https://github.com/rust-lang/crates.io-index"

8760checksum = "6bf85fd39652e740bf60f46f4cda9492c3a9ad75880575bf14960f775cb74a1c"

8761dependencies = [

8762 "cc",

8763 "tree-sitter-language",

8764]

8765

8766[[package]]

8767name = "tree-sitter-rust"

8768version = "0.24.2"

8769source = "registry+https://github.com/rust-lang/crates.io-index"

8770checksum = "439e577dbe07423ec2582ac62c7531120dbfccfa6e5f92406f93dd271a120e45"

8771dependencies = [

8772 "cc",

8773 "tree-sitter-language",

8774]

8775

8776[[package]]

8777name = "tree-sitter-typescript"

8778version = "0.23.2"

8779source = "registry+https://github.com/rust-lang/crates.io-index"

8780checksum = "6c5f76ed8d947a75cc446d5fccd8b602ebf0cde64ccf2ffa434d873d7a575eff"

8781dependencies = [

8782 "cc",

8783 "tree-sitter-language",

8784]

8785

8786[[package]]

86478787name = "try-lock"

86488788version = "0.2.5"

86498789source = "registry+https://github.com/rust-lang/crates.io-index"

@@ -9080,12 +9220,15 @@

90809220 "opentelemetry-otlp",

90819221 "opentelemetry_sdk",

90829222 "pulls",

9223 "search",

9224 "serde",

90839225 "session",

90849226 "space",

90859227 "tower-http 0.7.0",

90869228 "tracing",

90879229 "user",

90889230 "utoipa",

9231 "uuid",

90899232 "web",

90909233]

90919234

Modified Cargo.toml +11 −2 View Cargo.toml

@@ -29,6 +29,7 @@

2929git = { path = "lib/git" }

3030webapi = { path = "lib/webapi" }

3131pulls = { path = "lib/pulls" }

32search = { path = "lib/search" }

3233

3334

3435

@@ -81,11 +82,19 @@

8182maud = { version = "0.27", features = ["axum"] }

8283pulldown-cmark = { version = "0.13", default-features = false, features = ["html"] }

8384lettre = { version = "0.11", default-features = false, features = ["smtp-transport", "tokio1-rustls", "ring", "rustls-native-certs", "builder"] }

84reqwest = { version = "0.12", default-features = false, features = ["rustls-tls"] }

85reqwest = { version = "0.12", default-features = false, features = ["rustls-tls", "json"] }

8586anyhow = "1"

8687tracing-subscriber = { version = "0.3", features = ["env-filter"] }

8788jsonwebtoken = "9"

8889hyper = { version = "1", features = ["server", "http1"] }

8990hyper-util = { version = "0.1", features = ["tokio"] }

9091tokio-tungstenite = "0.29"

91

92tree-sitter = "0.25"

93tree-sitter-rust = "0.24.2"

94tree-sitter-javascript = "0.25.0"

95tree-sitter-typescript = "0.23.2"

96tree-sitter-python = "0.25.0"

97tree-sitter-go = "0.25.0"

98tree-sitter-java = "0.23.5"

99tree-sitter-c = "0.24.2"

100tree-sitter-cpp = "0.23.4"

Modified openapi.json View openapi.json

Diff too large to display.