feat(import): implement import service with CRUD operations and API endpoints

Modified app/synchost/src/main.rs +15 −0 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 import::ImportWorker;

910use search::{SearchService, SearchWorker};

1011use tokio::net::TcpListener;

1112use tokio_util::sync::CancellationToken;

@@ -49,6 +50,20 @@

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

5051 }

5152

53 if cfg.import.enabled {

54 let worker = ImportWorker::new(

55 db.clone(),

56 redis.clone(),

57 cache.clone(),

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

59 cfg.git.sync.clone(),

60 cfg.import.clone(),

61 cancel.clone(),

62 );

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

64 info!("import worker started");

65 }

66

5267 if cfg.health.enabled() {

5368 let health_addr: SocketAddr =

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

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

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

99export = { workspace = true }

1010config = { workspace = true }

1111cache = { workspace = true }

12import = { workspace = true }

1213search = { workspace = true }

1314axum = { workspace = true }

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

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

@@ -35,6 +35,9 @@

3535 APP_SEARCH__ENABLED: "false"

3636 APP_SEARCH__URL: "http://meilisearch:7700"

3737 APP_SEARCH__INDEX_PREFIX: "gitdata"

38 APP_IMPORT__ENABLED: "false"

39 # 自建 forge 的内网地址需显式加白，例如：

40 # APP_IMPORT__ALLOWED_HOSTS: '["gitea.internal","gitlab.svc.cluster.local"]'

3841

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

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

@@ -52,6 +55,7 @@

5255 APP_EMAIL__SMTP_PASSWORD: ""

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

5457 APP_SEARCH__API_KEY: ""

58 APP_IMPORT__TOKEN_SECRET: "" # >=32 字节，加密导入任务里的源站令牌

5559

5660podSecurityContext:

5761 fsGroup: 10001

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

@@ -43,6 +43,19 @@

4343poll_interval_secs = 2

4444task_timeout_secs = 120

4545

46[import]

47enabled = false

48token_secret = ""

49allowed_hosts = []

50allow_insecure_tls = false

51max_concurrent_jobs = 2

52clone_timeout_secs = 3600

53api_timeout_secs = 30

54poll_interval_secs = 3

55max_attempts = 5

56max_repo_bytes = 0

57stale_lock_secs = 1800

58

4659[database]

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

4861host = ""

Added lib/config/src/import.rs +77 −0 View lib/config/src/import.rs

@@ -0,0 +1,77 @@

1use serde::Deserialize;

2

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

4#[serde(default)]

5pub struct ImportConfig {

6 pub enabled: bool,

7 pub token_secret: String,

8 pub allowed_hosts: Vec<String>,

9 pub allow_insecure_tls: bool,

10 pub max_concurrent_jobs: usize,

11 pub clone_timeout_secs: u64,

12 pub api_timeout_secs: u64,

13 pub poll_interval_secs: u64,

14 pub max_attempts: i32,

15 pub max_repo_bytes: u64,

16 pub stale_lock_secs: u64,

17}

18

19impl Default for ImportConfig {

20 fn default() -> Self {

21 Self {

22 enabled: false,

23 token_secret: String::new(),

24 allowed_hosts: Vec::new(),

25 allow_insecure_tls: false,

26 max_concurrent_jobs: 2,

27 clone_timeout_secs: 3600,

28 api_timeout_secs: 30,

29 poll_interval_secs: 3,

30 max_attempts: 5,

31 max_repo_bytes: 0,

32 stale_lock_secs: 1800,

33 }

34 }

35}

36

37impl ImportConfig {

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

39 if !self.enabled {

40 return Ok(());

41 }

42 if self.token_secret.trim().is_empty() {

43 return Err("import.token_secret must not be empty when import is enabled".to_string());

44 }

45 if self.max_concurrent_jobs == 0

46 || self.clone_timeout_secs == 0

47 || self.api_timeout_secs == 0

48 || self.poll_interval_secs == 0

49 || self.stale_lock_secs == 0

50 {

51 return Err("import limits and intervals must be greater than zero".to_string());

52 }

53 if self.max_attempts <= 0 {

54 return Err("import.max_attempts must be greater than zero".to_string());

55 }

56 for host in &self.allowed_hosts {

57 let normalized = host.trim().to_ascii_lowercase();

58 if normalized.is_empty() || normalized.contains("://") || normalized.contains('/') {

59 return Err(format!(

60 "import.allowed_hosts entry `{host}` must be a bare host name"

61 ));

62 }

63 }

64 Ok(())

65 }

66

67 pub fn host_allowed(&self, host: &str) -> bool {

68 let host = host.trim().trim_end_matches('.').to_ascii_lowercase();

69 if host.is_empty() {

70 return false;

71 }

72 self.allowed_hosts.iter().any(|entry| {

73 let entry = entry.trim().trim_end_matches('.').to_ascii_lowercase();

74 !entry.is_empty() && (host == entry || host.ends_with(&format!(".{entry}")))

75 })

76 }

77}

Modified lib/config/src/lib.rs +4 −0 View lib/config/src/lib.rs

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

99pub mod githost;

1010pub mod health;

1111pub mod im;

12pub mod import;

1213pub mod issues;

1314pub mod pow;

1415pub mod quota;

@@ -37,6 +38,7 @@

3738use crate::githost::GithostConfig;

3839use crate::health::HealthConfig;

3940use crate::im::ImServerConfig;

41use crate::import::ImportConfig;

4042use crate::issues::IssueConfig;

4143use crate::pow::PowConfig;

4244use crate::redis::RedisConfig;

@@ -58,6 +60,7 @@

5860 pub database: DatabaseConfig,

5961 pub session: SessionConfig,

6062 pub search: SearchConfig,

63 pub import: ImportConfig,

6164 pub pow: PowConfig,

6265 pub auth: AuthConfig,

6366 pub web: WebConfig,

@@ -92,6 +95,7 @@

9295 .map_err(ConfigError::Validation)?;

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

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

98 self.import.validate().map_err(ConfigError::Validation)?;

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

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

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

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

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

1313use crate::githost::GithostConfig as GithostSection;

1414use crate::health::HealthConfig;

1515use crate::im::ImServerConfig;

16use crate::import::ImportConfig;

1617use crate::issues::IssueConfig;

1718use crate::pow::PowConfig;

1819use crate::quota::QuotaConfig;

@@ -74,6 +75,7 @@

7475 pub git: GitConfig,

7576 pub health: HealthConfig,

7677 pub search: SearchConfig,

78 pub import: ImportConfig,

7779}

7880

7981impl ScopedConfig for SynchostConfig {

@@ -82,6 +84,7 @@

8284 self.redis.validate()?;

8385 self.cache.validate()?;

8486 self.search.validate()?;

87 self.import.validate()?;

8588 self.health.is_valid()

8689 }

8790}

@@ -149,6 +152,7 @@

149152 pub email_queue: EmailQueueConfig,

150153 pub apihost: ApihostSection,

151154 pub search: SearchConfig,

155 pub import: ImportConfig,

152156}

153157

154158impl ApihostConfig {

@@ -172,6 +176,7 @@

172176 email_queue: self.email_queue,

173177 apihost: self.apihost,

174178 search: self.search,

179 import: self.import,

175180 ..Default::default()

176181 }

177182 }

@@ -189,6 +194,7 @@

189194 self.email_queue.is_valid()?;

190195 self.session.is_valid()?;

191196 self.search.validate()?;

197 self.import.validate()?;

192198 self.apihost.is_valid()

193199 }

194200}

Added lib/git/src/service/clone.rs +94 −0 View lib/git/src/service/clone.rs

@@ -0,0 +1,94 @@

1use std::path::Path;

2

3use crate::service::gitcmd::run_git;

4use crate::share::errors::GitError;

5

6#[derive(Debug, Clone)]

7pub struct GitCredential {

8 pub username: String,

9 pub secret: String,

10}

11

12impl GitCredential {

13 #[must_use]

14 pub fn new(username: impl Into<String>, secret: impl Into<String>) -> Self {

15 Self {

16 username: username.into(),

17 secret: secret.into(),

18 }

19 }

20}

21

22const ASKPASS_SCRIPT: &str = r#"#!/bin/sh

23case "$1" in

24 *sername*) printf '%s\n' "$GITDATA_GIT_USER" ;;

25 *) printf '%s\n' "$GITDATA_GIT_PASS" ;;

26esac

27"#;

28

29const MIRROR_REFSPECS: [&str; 2] = ["+refs/heads/*:refs/heads/*", "+refs/tags/*:refs/tags/*"];

30

31fn mirror_args(url: &str) -> Vec<String> {

32 let mut args = vec![

33 "fetch".to_string(),

34 "--prune".to_string(),

35 "--force".to_string(),

36 "--".to_string(),

37 url.to_string(),

38 ];

39 args.extend(MIRROR_REFSPECS.iter().map(|spec| (*spec).to_string()));

40 args

41}

42

43pub async fn mirror_fetch(

44 label: &str,

45 dir: &Path,

46 url: &str,

47 credential: Option<&GitCredential>,

48 timeout_secs: u64,

49) -> Result<(), GitError> {

50 let Some(cred) = credential else {

51 run_git(label, dir, mirror_args(url), &[], timeout_secs).await?;

52 return Ok(());

53 };

54

55 let script = std::env::temp_dir().join(format!("gitdata-askpass-{}.sh", uuid::Uuid::now_v7()));

56 tokio::fs::write(&script, ASKPASS_SCRIPT)

57 .await

58 .map_err(|e| GitError::Internal(format!("write askpass script: {e}")))?;

59 if let Err(e) = set_executable(&script).await {

60 let _ = tokio::fs::remove_file(&script).await;

61 return Err(e);

62 }

63

64 let env = vec![

65 (

66 "GIT_ASKPASS".to_string(),

67 script.to_string_lossy().to_string(),

68 ),

69 ("GITDATA_GIT_USER".to_string(), cred.username.clone()),

70 ("GITDATA_GIT_PASS".to_string(), cred.secret.clone()),

71 ("GIT_TERMINAL_PROMPT".to_string(), "0".to_string()),

72 ];

73 let result = run_git(label, dir, mirror_args(url), &env, timeout_secs).await;

74 if let Err(e) = tokio::fs::remove_file(&script).await {

75 tracing::warn!(error = %e, "failed to remove askpass script");

76 }

77 result?;

78 Ok(())

79}

80

81#[cfg(unix)]

82async fn set_executable(path: &Path) -> Result<(), GitError> {

83 use std::os::unix::fs::PermissionsExt;

84 let perms = std::fs::Permissions::from_mode(0o700);

85 tokio::fs::set_permissions(path, perms)

86 .await

87 .map_err(|e| GitError::Internal(format!("chmod askpass script: {e}")))?;

88 Ok(())

89}

90

91#[cfg(not(unix))]

92async fn set_executable(_path: &Path) -> Result<(), GitError> {

93 Ok(())

94}

Modified lib/git/src/service/mod.rs +1 −0 View lib/git/src/service/mod.rs

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

3535pub mod blobs;

3636pub mod branches;

3737pub mod cache;

38pub mod clone;

3839pub mod commits;

3940pub mod conflicts;

4041pub mod content;

Modified lib/git/src/service/remotes.rs +18 −9 View lib/git/src/service/remotes.rs

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

11use serde::{Deserialize, Serialize};

22

33use crate::service::GitService;

4use crate::service::clone::{GitCredential, mirror_fetch};

45use crate::service::gitcmd::{run_git, run_git_soft};

56use crate::share::errors::{GitError, map_baby};

67use crate::share::validate::{validate_refspec, validate_remote_name, validate_remote_url};

@@ -27,6 +28,8 @@

2728pub struct RemoteAuthDto {

2829 #[serde(default)]

2930 pub token: Option<String>,

31 #[serde(default)]

32 pub username: Option<String>,

3033}

3134

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

@@ -46,20 +49,26 @@

4649 space_name: String,

4750 repo_name: String,

4851 url: String,

49 _auth: Option<RemoteAuthDto>,

52 auth: Option<RemoteAuthDto>,

5053 ) -> Result<ImportResultDto, GitError> {

5154 validate_remote_url(&url)?;

5255 let ctx = self.context(space_name, repo_name).await?;

5356 let dir = ctx.repo_dir();

5457 let timeout = self.state.config.git.git_command_timeout_secs;

55 run_git(

56 "remotes::import",

57 &dir,

58 vec!["fetch".to_string(), "--mirror".to_string(), url],

59 &[],

60 timeout,

61 )

62 .await?;

58 let credential = auth

59 .as_ref()

60 .and_then(|a| a.token.clone())

61 .filter(|t| !t.is_empty())

62 .map(|token| GitCredential {

63 username: auth

64 .as_ref()

65 .and_then(|a| a.username.clone())

66 .map(|u| u.trim().to_string())

67 .filter(|u| !u.is_empty())

68 .unwrap_or_else(|| "git".to_string()),

69 secret: token,

70 });

71 mirror_fetch("remotes::import", &dir, &url, credential.as_ref(), timeout).await?;

6372 let baby = self.baby(&ctx);

6473 let default = baby.get_default_branch().await.map_err(map_baby)?;

6574 let commit_count = count_refs(&dir).await?;

Modified lib/git/src/sync/commits.rs +22 −2 View lib/git/src/sync/commits.rs

@@ -4,7 +4,7 @@

44use gitbaby::GitBaby;

55use gitbaby::commit::types::ListCommitsOptions;

66use models::git::git_commits;

7use models::users::user_emails;

7use models::users::{user_emails, user_external_identity};

88use sea_orm::{ColumnTrait, DatabaseTransaction, EntityTrait, QueryFilter, QuerySelect, Set};

99use time::OffsetDateTime;

1010use uuid::Uuid;

@@ -174,10 +174,30 @@

174174 .await

175175 .map_err(SyncError::Db)?;

176176

177 let map: HashMap<String, Uuid> = rows

177 let mut map: HashMap<String, Uuid> = rows

178178 .into_iter()

179179 .map(|r| (r.lower_email, r.user_id))

180180 .collect();

181

182 let unmatched: Vec<String> = lower_emails

183 .iter()

184 .filter(|email| !map.contains_key(*email))

185 .cloned()

186 .collect();

187 if !unmatched.is_empty() {

188 let identities = user_external_identity::Entity::find()

189 .filter(user_external_identity::Column::UserId.is_not_null())

190 .filter(user_external_identity::Column::Email.is_in(unmatched))

191 .all(txn)

192 .await

193 .map_err(SyncError::Db)?;

194 for row in identities {

195 if let (Some(email), Some(uid)) = (row.email, row.user_id) {

196 map.entry(email.to_lowercase()).or_insert(uid);

197 }

198 }

199 }

200

181201 tracing::debug!(

182202 emails = lower_emails.len(),

183203 matched = map.len(),

Added lib/import/src/handlers/cancel_import.rs +30 −0 View lib/import/src/handlers/cancel_import.rs

@@ -0,0 +1,30 @@

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

2use export::rest::AppResponseData;

3use session::Session;

4

5use crate::errors::ApiError;

6use crate::handlers::caller;

7use crate::service::{ImportJobDto, ImportJobPath, ImportService};

8

9#[utoipa::path(

10 post,

11 path = "/imports/{space_name}/{job_id}/cancel",

12 tag = "imports",

13 description = "取消排队中或执行中的导入任务；worker 在阶段边界停止并清除已保存的令牌。",

14 params(ImportJobPath),

15 responses(

16 (status = 200, description = "cancel import job", body = AppResponseData<ImportJobDto>)

17 )

18)]

19pub async fn cancel_import(

20 State(service): State<ImportService>,

21 session: Session,

22 Path(path): Path<ImportJobPath>,

23) -> Result<AppResponseData<ImportJobDto>, ApiError> {

24 let caller = caller(&session)?;

25 Ok(AppResponseData::ok(

26 service

27 .cancel(caller, &path.space_name, path.job_id)

28 .await?,

29 ))

30}

Added lib/import/src/handlers/create_import.rs +31 −0 View lib/import/src/handlers/create_import.rs

@@ -0,0 +1,31 @@

1use axum::Json;

2use axum::extract::{Path, State};

3use export::rest::AppResponseData;

4use session::Session;

5

6use crate::errors::ApiError;

7use crate::handlers::caller;

8use crate::service::{CreateImportCmd, ImportJobDto, ImportService, ImportSpacePath};

9

10#[utoipa::path(

11 post,

12 path = "/imports/{space_name}",

13 tag = "imports",

14 description = "创建导入任务：先探测源仓库并在空间下创建目标仓库，再由 synchost 异步镜像抓取全部引用并同步元数据；仅空间所有者可操作。令牌加密保存在任务上，任务结束即清除。",

15 params(ImportSpacePath),

16 request_body = CreateImportCmd,

17 responses(

18 (status = 200, description = "create import job", body = AppResponseData<ImportJobDto>)

19 )

20)]

21pub async fn create_import(

22 State(service): State<ImportService>,

23 session: Session,

24 Path(path): Path<ImportSpacePath>,

25 Json(cmd): Json<CreateImportCmd>,

26) -> Result<AppResponseData<ImportJobDto>, ApiError> {

27 let caller = caller(&session)?;

28 Ok(AppResponseData::ok(

29 service.create(caller, &path.space_name, cmd).await?,

30 ))

31}

Added lib/import/src/handlers/get_import.rs +28 −0 View lib/import/src/handlers/get_import.rs

@@ -0,0 +1,28 @@

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

2use export::rest::AppResponseData;

3use session::Session;

4

5use crate::errors::ApiError;

6use crate::handlers::caller;

7use crate::service::{ImportJobDto, ImportJobPath, ImportService};

8

9#[utoipa::path(

10 get,

11 path = "/imports/{space_name}/{job_id}",

12 tag = "imports",

13 description = "查询单个导入任务的状态、阶段、统计与失败原因。",

14 params(ImportJobPath),

15 responses(

16 (status = 200, description = "get import job", body = AppResponseData<ImportJobDto>)

17 )

18)]

19pub async fn get_import(

20 State(service): State<ImportService>,

21 session: Session,

22 Path(path): Path<ImportJobPath>,

23) -> Result<AppResponseData<ImportJobDto>, ApiError> {

24 let caller = caller(&session)?;

25 Ok(AppResponseData::ok(

26 service.get(caller, &path.space_name, path.job_id).await?,

27 ))

28}

Added lib/import/src/handlers/list_imports.rs +29 −0 View lib/import/src/handlers/list_imports.rs

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

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

2use export::rest::AppResponseData;

3use session::Session;

4

5use crate::errors::ApiError;

6use crate::handlers::caller;

7use crate::service::{ImportJobPage, ImportService, ImportSpacePath, ListImportQuery};

8

9#[utoipa::path(

10 get,

11 path = "/imports/{space_name}",

12 tag = "imports",

13 description = "分页列出空间内的导入任务；空间所有者与平台管理员可见全部，其余成员仅见自己发起的任务。",

14 params(ImportSpacePath, ListImportQuery),

15 responses(

16 (status = 200, description = "list import jobs", body = AppResponseData<ImportJobPage>)

17 )

18)]

19pub async fn list_imports(

20 State(service): State<ImportService>,

21 session: Session,

22 Path(path): Path<ImportSpacePath>,

23 Query(query): Query<ListImportQuery>,

24) -> Result<AppResponseData<ImportJobPage>, ApiError> {

25 let caller = caller(&session)?;

26 Ok(AppResponseData::ok(

27 service.list(caller, &path.space_name, query).await?,

28 ))

29}

Added lib/import/src/handlers/mod.rs +16 −0 View lib/import/src/handlers/mod.rs

@@ -0,0 +1,16 @@

1pub mod cancel_import;

2pub mod create_import;

3pub mod get_import;

4pub mod list_imports;

5pub mod probe_import;

6pub mod requeue_imports;

7pub mod retry_import;

8

9use session::{Session, SessionUser};

10use uuid::Uuid;

11

12use crate::errors::ApiError;

13

14pub fn caller(session: &Session) -> Result<Uuid, ApiError> {

15 session.uid().ok_or_else(ApiError::unauthorized)

16}

Added lib/import/src/handlers/probe_import.rs +31 −0 View lib/import/src/handlers/probe_import.rs

@@ -0,0 +1,31 @@

1use axum::Json;

2use axum::extract::{Path, State};

3use export::rest::AppResponseData;

4use session::Session;

5

6use crate::errors::ApiError;

7use crate::handlers::caller;

8use crate::service::{ImportProbeDto, ImportService, ImportSpacePath, ProbeImportCmd};

9

10#[utoipa::path(

11 post,

12 path = "/imports/{space_name}/probe",

13 tag = "imports",

14 description = "探测 GitHub / GitLab / Gitea 源仓库，返回名称、默认分支、私有性与体积预览，不落库也不创建仓库。",

15 params(ImportSpacePath),

16 request_body = ProbeImportCmd,

17 responses(

18 (status = 200, description = "probe import source", body = AppResponseData<ImportProbeDto>)

19 )

20)]

21pub async fn probe_import(

22 State(service): State<ImportService>,

23 session: Session,

24 Path(path): Path<ImportSpacePath>,

25 Json(cmd): Json<ProbeImportCmd>,

26) -> Result<AppResponseData<ImportProbeDto>, ApiError> {

27 let caller = caller(&session)?;

28 Ok(AppResponseData::ok(

29 service.probe(caller, &path.space_name, cmd).await?,

30 ))

31}

Added lib/import/src/handlers/requeue_imports.rs +11 −0 View lib/import/src/handlers/requeue_imports.rs

@@ -0,0 +1,11 @@

1use axum::extract::State;

2use export::rest::AppResponseData;

3

4use crate::errors::ApiError;

5use crate::service::{ImportService, RequeueResultDto};

6

7pub async fn requeue_imports(

8 State(service): State<ImportService>,

9) -> Result<AppResponseData<RequeueResultDto>, ApiError> {

10 Ok(AppResponseData::ok(service.requeue_pending().await?))

11}

Added lib/import/src/handlers/retry_import.rs +33 −0 View lib/import/src/handlers/retry_import.rs

@@ -0,0 +1,33 @@

1use axum::Json;

2use axum::extract::{Path, State};

3use export::rest::AppResponseData;

4use session::Session;

5

6use crate::errors::ApiError;

7use crate::handlers::caller;

8use crate::service::{ImportJobDto, ImportJobPath, ImportService, RetryImportCmd};

9

10#[utoipa::path(

11 post,

12 path = "/imports/{space_name}/{job_id}/retry",

13 tag = "imports",

14 description = "重试已失败或已取消的导入任务；令牌在终态已被清除，私有仓库需在此重新提供。",

15 params(ImportJobPath),

16 request_body = RetryImportCmd,

17 responses(

18 (status = 200, description = "retry import job", body = AppResponseData<ImportJobDto>)

19 )

20)]

21pub async fn retry_import(

22 State(service): State<ImportService>,

23 session: Session,

24 Path(path): Path<ImportJobPath>,

25 Json(cmd): Json<RetryImportCmd>,

26) -> Result<AppResponseData<ImportJobDto>, ApiError> {

27 let caller = caller(&session)?;

28 Ok(AppResponseData::ok(

29 service

30 .retry(caller, &path.space_name, path.job_id, cmd)

31 .await?,

32 ))

33}

Added lib/import/src/providers/gitea.rs +48 −0 View lib/import/src/providers/gitea.rs

@@ -0,0 +1,48 @@

1use crate::errors::ImportError;

2

3use super::{

4 ImportProvider, ProbeInfo, ProbeRequest, api_base, get_json, http_client, json_bool, json_i64,

5 json_string, source_path,

6};

7

8pub async fn probe(request: ProbeRequest<'_>) -> Result<ProbeInfo, ImportError> {

9 let client = http_client(request.config)?;

10 let base = api_base(ImportProvider::Gitea, request.base_url, request.config)?;

11 let (path, fallback_name) = source_path(ImportProvider::Gitea, request.source)?;

12 let url = format!("{base}/repos/{path}");

13 let value = get_json(&client, &url, ImportProvider::Gitea, request.token).await?;

14 let clone_url = json_string(&value, &["clone_url"]).ok_or_else(|| {

15 ImportError::Upstream("source forge did not return a clone url".to_string())

16 })?;

17 let owner = json_string(&value, &["owner", "login"])

18 .unwrap_or_else(|| path.split('/').next().unwrap_or_default().to_string());

19 let credential_username = match request.token.filter(|token| !token.trim().is_empty()) {

20 Some(token) => {

21 let me = get_json(

22 &client,

23 &format!("{base}/user"),

24 ImportProvider::Gitea,

25 Some(token),

26 )

27 .await

28 .ok()

29 .and_then(|value| json_string(&value, &["login"]))

30 .filter(|login| !login.is_empty());

31 Some(me.unwrap_or_else(|| owner.clone()))

32 }

33 None => None,

34 };

35 Ok(ProbeInfo {

36 provider: ImportProvider::Gitea,

37 owner,

38 name: json_string(&value, &["name"]).unwrap_or(fallback_name),

39 clone_url,

40 default_branch: json_string(&value, &["default_branch"]).unwrap_or_else(|| "main".into()),

41 is_private: json_bool(&value, &["private"]).unwrap_or(false),

42 description: json_string(&value, &["description"]),

43 size_bytes: json_i64(&value, &["size"]).map(|kb| kb.saturating_mul(1024)),

44 owner_external_id: json_string(&value, &["owner", "id"]),

45 owner_avatar_url: json_string(&value, &["owner", "avatar_url"]),

46 credential_username,

47 })

48}

Added lib/import/src/providers/github.rs +37 −0 View lib/import/src/providers/github.rs

@@ -0,0 +1,37 @@

1use crate::errors::ImportError;

2

3use super::{

4 ImportProvider, ProbeInfo, ProbeRequest, api_base, get_json, http_client, json_bool, json_i64,

5 json_string, source_path,

6};

7

8pub async fn probe(request: ProbeRequest<'_>) -> Result<ProbeInfo, ImportError> {

9 let client = http_client(request.config)?;

10 let base = api_base(ImportProvider::Github, request.base_url, request.config)?;

11 let (path, fallback_name) = source_path(ImportProvider::Github, request.source)?;

12 let url = format!("{base}/repos/{path}");

13 let value = get_json(&client, &url, ImportProvider::Github, request.token).await?;

14 let clone_url = json_string(&value, &["clone_url"]).ok_or_else(|| {

15 ImportError::Upstream("source forge did not return a clone url".to_string())

16 })?;

17 let owner = json_string(&value, &["owner", "login"])

18 .unwrap_or_else(|| path.split('/').next().unwrap_or_default().to_string());

19 let has_token = request.token.is_some_and(|token| !token.trim().is_empty());

20 Ok(ProbeInfo {

21 provider: ImportProvider::Github,

22 owner,

23 name: json_string(&value, &["name"]).unwrap_or(fallback_name),

24 clone_url,

25 default_branch: json_string(&value, &["default_branch"]).unwrap_or_else(|| "main".into()),

26 is_private: json_bool(&value, &["private"]).unwrap_or(false),

27 description: json_string(&value, &["description"]),

28 size_bytes: json_i64(&value, &["size"]).map(|kb| kb.saturating_mul(1024)),

29 owner_external_id: json_string(&value, &["owner", "id"]),

30 owner_avatar_url: json_string(&value, &["owner", "avatar_url"]),

31 credential_username: has_token.then(|| {

32 ImportProvider::Github

33 .default_credential_username()

34 .to_string()

35 }),

36 })

37}

Added lib/import/src/providers/gitlab.rs +44 −0 View lib/import/src/providers/gitlab.rs

@@ -0,0 +1,44 @@

1use crate::errors::ImportError;

2

3use super::{

4 ImportProvider, ProbeInfo, ProbeRequest, api_base, encode_project_path, get_json, http_client,

5 json_i64, json_string, source_path,

6};

7

8pub async fn probe(request: ProbeRequest<'_>) -> Result<ProbeInfo, ImportError> {

9 let client = http_client(request.config)?;

10 let base = api_base(ImportProvider::Gitlab, request.base_url, request.config)?;

11 let (path, fallback_name) = source_path(ImportProvider::Gitlab, request.source)?;

12 let url = format!(

13 "{base}/projects/{}?statistics=true",

14 encode_project_path(&path)

15 );

16 let value = get_json(&client, &url, ImportProvider::Gitlab, request.token).await?;

17 let clone_url = json_string(&value, &["http_url_to_repo"]).ok_or_else(|| {

18 ImportError::Upstream("source forge did not return a clone url".to_string())

19 })?;

20 let owner = json_string(&value, &["namespace", "full_path"])

21 .or_else(|| json_string(&value, &["namespace", "path"]))

22 .unwrap_or_else(|| path.split('/').next().unwrap_or_default().to_string());

23 let visibility = json_string(&value, &["visibility"]).unwrap_or_default();

24 let has_token = request.token.is_some_and(|token| !token.trim().is_empty());

25 Ok(ProbeInfo {

26 provider: ImportProvider::Gitlab,

27 owner,

28 name: json_string(&value, &["name"])

29 .or_else(|| json_string(&value, &["path"]))

30 .unwrap_or(fallback_name),

31 clone_url,

32 default_branch: json_string(&value, &["default_branch"]).unwrap_or_else(|| "main".into()),

33 is_private: visibility == "private",

34 description: json_string(&value, &["description"]),

35 size_bytes: json_i64(&value, &["statistics", "disk_size_bytes"]),

36 owner_external_id: json_string(&value, &["namespace", "id"]),

37 owner_avatar_url: json_string(&value, &["namespace", "avatar_url"]),

38 credential_username: has_token.then(|| {

39 ImportProvider::Gitlab

40 .default_credential_username()

41 .to_string()

42 }),

43 })

44}

Added lib/import/src/providers/mod.rs +326 −0 View lib/import/src/providers/mod.rs

@@ -0,0 +1,326 @@

1pub mod gitea;

2pub mod github;

3pub mod gitlab;

4

5use std::time::Duration;

6

7use config::import::ImportConfig;

8use reqwest::{Client, header};

9use serde::{Deserialize, Serialize};

10use utoipa::ToSchema;

11

12use crate::errors::ImportError;

13

14const MAX_RETRY_ATTEMPTS: u32 = 2;

15

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

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

18pub enum ImportProvider {

19 Github,

20 Gitlab,

21 Gitea,

22}

23

24impl ImportProvider {

25 #[must_use]

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

27 match self {

28 Self::Github => "github",

29 Self::Gitlab => "gitlab",

30 Self::Gitea => "gitea",

31 }

32 }

33

34 pub fn parse(value: &str) -> Result<Self, ImportError> {

35 match value.trim().to_ascii_lowercase().as_str() {

36 "github" => Ok(Self::Github),

37 "gitlab" => Ok(Self::Gitlab),

38 "gitea" => Ok(Self::Gitea),

39 other => Err(ImportError::Internal(format!(

40 "unknown import provider `{other}`"

41 ))),

42 }

43 }

44

45 #[must_use]

46 pub fn default_root(self) -> &'static str {

47 match self {

48 Self::Github => "https://github.com",

49 Self::Gitlab => "https://gitlab.com",

50 Self::Gitea => "https://gitea.com",

51 }

52 }

53

54 #[must_use]

55 pub fn default_credential_username(self) -> &'static str {

56 match self {

57 Self::Github => "x-access-token",

58 Self::Gitlab => "oauth2",

59 Self::Gitea => "git",

60 }

61 }

62}

63

64#[derive(Debug, Clone, Serialize, ToSchema)]

65pub struct ProbeInfo {

66 pub provider: ImportProvider,

67 pub owner: String,

68 pub name: String,

69 pub clone_url: String,

70 pub default_branch: String,

71 pub is_private: bool,

72 pub description: Option<String>,

73 pub size_bytes: Option<i64>,

74 pub owner_external_id: Option<String>,

75 pub owner_avatar_url: Option<String>,

76 pub credential_username: Option<String>,

77}

78

79pub struct ProbeRequest<'a> {

80 pub provider: ImportProvider,

81 pub source: &'a str,

82 pub base_url: Option<&'a str>,

83 pub token: Option<&'a str>,

84 pub config: &'a ImportConfig,

85}

86

87pub async fn probe(request: ProbeRequest<'_>) -> Result<ProbeInfo, ImportError> {

88 match request.provider {

89 ImportProvider::Github => github::probe(request).await,

90 ImportProvider::Gitlab => gitlab::probe(request).await,

91 ImportProvider::Gitea => gitea::probe(request).await,

92 }

93}

94

95pub(crate) fn http_client(config: &ImportConfig) -> Result<Client, ImportError> {

96 Client::builder()

97 .timeout(Duration::from_secs(config.api_timeout_secs.max(1)))

98 .danger_accept_invalid_certs(config.allow_insecure_tls)

99 .user_agent("gitdata-import/1.0")

100 .build()

101 .map_err(|e| ImportError::Internal(format!("build import http client: {e}")))

102}

103

104pub(crate) fn host_of(url: &str) -> Option<String> {

105 let rest = url.split_once("://")?.1;

106 let authority = rest.split(['/', '?', '#']).next()?;

107 let without_user = authority.rsplit('@').next()?;

108 let host = strip_brackets_and_port(without_user);

109 if host.is_empty() {

110 None

111 } else {

112 Some(host.to_ascii_lowercase())

113 }

114}

115

116fn strip_brackets_and_port(host: &str) -> String {

117 if host.starts_with('[') {

118 if let Some(end) = host.find(']') {

119 return host[1..end].to_string();

120 }

121 }

122 if host.matches(':').count() == 1

123 && let Some(idx) = host.rfind(':')

124 {

125 let (h, port) = host.split_at(idx);

126 let port = &port[1..];

127 if !port.is_empty() && port.chars().all(|c| c.is_ascii_digit()) {

128 return h.to_string();

129 }

130 }

131 host.to_string()

132}

133

134pub fn ensure_url_allowed(url: &str, config: &ImportConfig) -> Result<(), ImportError> {

135 let trimmed = url.trim();

136 if trimmed.is_empty() || trimmed.starts_with('-') {

137 return Err(ImportError::BadRequest("source url is empty".to_string()));

138 }

139 if trimmed.chars().any(|c| c.is_control() || c.is_whitespace()) {

140 return Err(ImportError::BadRequest(

141 "source url contains control characters".to_string(),

142 ));

143 }

144 let Some((scheme, rest)) = trimmed.split_once("://") else {

145 return Err(ImportError::BadRequest(format!(

146 "source url `{trimmed}` must include http:// or https://"

147 )));

148 };

149 let scheme = scheme.to_ascii_lowercase();

150 if scheme != "http" && scheme != "https" {

151 return Err(ImportError::BadRequest(format!(

152 "source url scheme `{scheme}` is not allowed, use http or https"

153 )));

154 }

155 if let Some(userinfo) = rest

156 .split(['/', '?', '#'])

157 .next()

158 .and_then(|a| a.rsplit_once('@'))

159 && userinfo.0.contains(':')

160 {

161 return Err(ImportError::BadRequest(

162 "source url must not embed credentials".to_string(),

163 ));

164 }

165 let Some(host) = host_of(trimmed) else {

166 return Err(ImportError::BadRequest(format!(

167 "source url `{trimmed}` has no host"

168 )));

169 };

170 if git::share::validate::validate_remote_url(trimmed).is_err() && !config.host_allowed(&host) {

171 return Err(ImportError::BadRequest(format!(

172 "source host `{host}` is not allowed; add it to import.allowed_hosts for self-hosted forges"

173 )));

174 }

175 Ok(())

176}

177

178pub(crate) fn api_base(

179 provider: ImportProvider,

180 base_url: Option<&str>,

181 config: &ImportConfig,

182) -> Result<String, ImportError> {

183 let root = match base_url.map(str::trim).filter(|value| !value.is_empty()) {

184 Some(value) => value.trim_end_matches('/').to_string(),

185 None => provider.default_root().to_string(),

186 };

187 ensure_url_allowed(&root, config)?;

188 let host = host_of(&root).unwrap_or_default();

189 Ok(match provider {

190 ImportProvider::Github if host == "github.com" || host == "api.github.com" => {

191 "https://api.github.com".to_string()

192 }

193 ImportProvider::Github => format!("{root}/api/v3"),

194 ImportProvider::Gitlab => format!("{root}/api/v4"),

195 ImportProvider::Gitea => format!("{root}/api/v1"),

196 })

197}

198

199pub(crate) fn source_path(

200 provider: ImportProvider,

201 source: &str,

202) -> Result<(String, String), ImportError> {

203 let trimmed = source.trim().trim_end_matches('/');

204 if trimmed.is_empty() {

205 return Err(ImportError::BadRequest("source is empty".to_string()));

206 }

207 let without_host = match trimmed.split_once("://") {

208 Some((_, rest)) => match rest.split_once('/') {

209 Some((_, path)) => path,

210 None => "",

211 },

212 None => trimmed,

213 };

214 let path = without_host.split(['?', '#']).next().unwrap_or_default();

215 let path = path.trim_matches('/');

216 let path = path.strip_suffix(".git").unwrap_or(path);

217 let segments: Vec<&str> = path.split('/').filter(|s| !s.is_empty()).collect();

218 if segments.len() < 2 {

219 return Err(ImportError::BadRequest(format!(

220 "cannot read owner and repository from source `{source}`; use `owner/repo` or a full repository url"

221 )));

222 }

223 let max_segments = match provider {

224 ImportProvider::Gitlab => 20,

225 _ => 2,

226 };

227 if segments.len() > max_segments {

228 return Err(ImportError::BadRequest(format!(

229 "source `{source}` has too many path segments for {}",

230 provider.as_str()

231 )));

232 }

233 if segments

234 .iter()

235 .any(|s| s.chars().any(|c| c.is_control() || c.is_whitespace()))

236 {

237 return Err(ImportError::BadRequest(format!(

238 "source `{source}` contains invalid characters"

239 )));

240 }

241 let full = segments.join("/");

242 let name = segments.last().copied().unwrap_or_default().to_string();

243 Ok((full, name))

244}

245

246pub(crate) fn encode_project_path(path: &str) -> String {

247 path.replace('/', "%2F")

248}

249

250pub(crate) async fn get_json(

251 client: &Client,

252 url: &str,

253 provider: ImportProvider,

254 token: Option<&str>,

255) -> Result<serde_json::Value, ImportError> {

256 let mut attempt: u32 = 0;

257 loop {

258 let mut request = client.get(url).header(header::ACCEPT, "application/json");

259 if let Some(token) = token.filter(|value| !value.trim().is_empty()) {

260 request = match provider {

261 ImportProvider::Github => request

262 .header(header::AUTHORIZATION, format!("Bearer {token}"))

263 .header("X-GitHub-Api-Version", "2022-11-28"),

264 ImportProvider::Gitlab => request.header("PRIVATE-TOKEN", token),

265 ImportProvider::Gitea => {

266 request.header(header::AUTHORIZATION, format!("token {token}"))

267 }

268 };

269 }

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

271 let status = response.status();

272 if status.is_success() {

273 let text = response.text().await?;

274 return serde_json::from_str(&text).map_err(|e| {

275 ImportError::Upstream(format!("source forge returned invalid json: {e}"))

276 });

277 }

278 let body = response.text().await.unwrap_or_default();

279 let detail: String = body.chars().take(300).collect();

280 let code = status.as_u16();

281 if code == 401 || code == 403 {

282 return Err(ImportError::BadRequest(format!(

283 "source forge rejected the credentials ({code}): {detail}"

284 )));

285 }

286 if code == 404 {

287 return Err(ImportError::NotFound);

288 }

289 if (code == 429 || (500..=599).contains(&code)) && attempt < MAX_RETRY_ATTEMPTS {

290 attempt += 1;

291 tokio::time::sleep(Duration::from_secs(u64::from(attempt) * 2)).await;

292 continue;

293 }

294 return Err(ImportError::Upstream(format!(

295 "source forge returned {code}: {detail}"

296 )));

297 }

298}

299

300pub(crate) fn json_string(value: &serde_json::Value, path: &[&str]) -> Option<String> {

301 let mut current = value;

302 for key in path {

303 current = current.get(key)?;

304 }

305 match current {

306 serde_json::Value::String(text) if !text.is_empty() => Some(text.clone()),

307 serde_json::Value::Number(number) => Some(number.to_string()),

308 _ => None,

309 }

310}

311

312pub(crate) fn json_bool(value: &serde_json::Value, path: &[&str]) -> Option<bool> {

313 let mut current = value;

314 for key in path {

315 current = current.get(key)?;

316 }

317 current.as_bool()

318}

319

320pub(crate) fn json_i64(value: &serde_json::Value, path: &[&str]) -> Option<i64> {

321 let mut current = value;

322 for key in path {

323 current = current.get(key)?;

324 }

325 current.as_i64()

326}

Added lib/import/src/share/guard.rs +16 −0 View lib/import/src/share/guard.rs

@@ -0,0 +1,16 @@

1use axum::extract::{Request, State};

2use axum::middleware::Next;

3use axum::response::Response;

4use session::Session;

5use session::share::guard::session_guard;

6

7use crate::service::ImportService;

8

9pub async fn import_guard(

10 State(service): State<ImportService>,

11 session: Session,

12 request: Request,

13 next: Next,

14) -> Response {

15 session_guard::<ImportService>(State(service), session, request, next).await

16}

Added lib/import/src/share/mod.rs +1 −0 View lib/import/src/share/mod.rs

@@ -0,0 +1 @@

1pub mod guard;

Added lib/import/src/errors.rs +115 −0 View lib/import/src/errors.rs

@@ -0,0 +1,115 @@

1use std::fmt;

2

3#[derive(Debug)]

4pub enum ImportError {

5 Disabled,

6 Canceled,

7 BadRequest(String),

8 Unauthorized,

9 Forbidden,

10 NotFound,

11 Conflict(String),

12 RateLimited(String),

13 Upstream(String),

14 Internal(String),

15}

16

17impl ImportError {

18 #[must_use]

19 pub fn is_permanent(&self) -> bool {

20 matches!(

21 self,

22 Self::Disabled

23 | Self::Canceled

24 | Self::BadRequest(_)

25 | Self::Unauthorized

26 | Self::Forbidden

27 | Self::NotFound

28 | Self::Conflict(_)

29 )

30 }

31}

32

33impl fmt::Display for ImportError {

34 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {

35 match self {

36 Self::Disabled => write!(f, "import is disabled"),

37 Self::Canceled => write!(f, "import canceled"),

38 Self::BadRequest(message) => write!(f, "{message}"),

39 Self::Unauthorized => write!(f, "unauthorized"),

40 Self::Forbidden => write!(f, "forbidden"),

41 Self::NotFound => write!(f, "not found"),

42 Self::Conflict(message) => write!(f, "{message}"),

43 Self::RateLimited(message) => write!(f, "{message}"),

44 Self::Upstream(message) => write!(f, "{message}"),

45 Self::Internal(message) => write!(f, "{message}"),

46 }

47 }

48}

49

50impl std::error::Error for ImportError {}

51

52export::db_error_conversions!(ImportError);

53export::export_error_conversion!(ImportError);

54export::json_payload_conversion!(ImportError);

55

56impl From<git::share::errors::GitError> for ImportError {

57 fn from(error: git::share::errors::GitError) -> Self {

58 use git::share::errors::GitError as G;

59 match error {

60 G::InvalidArgument(msg) => Self::BadRequest(msg),

61 G::NotFound => Self::NotFound,

62 G::Unauthorized => Self::Unauthorized,

63 G::Forbidden => Self::Forbidden,

64 G::Conflict(msg) => Self::Conflict(msg),

65 G::Cache(_) | G::Git(_) | G::Internal(_) => Self::Internal(error.to_string()),

66 }

67 }

68}

69

70impl From<reqwest::Error> for ImportError {

71 fn from(error: reqwest::Error) -> Self {

72 Self::Upstream(format!("source forge request failed: {error}"))

73 }

74}

75

76export::api_error! {

77 log: "import",

78 pub enum ApiError {

79 BadRequest(400, "request failed validation"),

80 Unauthorized(401, "authentication required"),

81 Forbidden(403, "operation not allowed"),

82 NotFound(404, "resource not found"),

83 Conflict(409, "resource conflict"),

84 TooManyRequests(429, "rate limit exceeded"),

85 Internal(500, "internal server error"),

86 BadGateway(502, "upstream forge request failed"),

87 ServiceUnavailable(503, "import is unavailable"),

88 }

89}

90

91impl From<ImportError> for ApiError {

92 fn from(error: ImportError) -> Self {

93 match error {

94 ImportError::Disabled => Self::build(

95 503,

96 export::rest::AppResponseData::err(503, "import is unavailable")

97 .with_error_code("import_disabled"),

98 ),

99 ImportError::Canceled => Self::bad_request("import job canceled"),

100 ImportError::BadRequest(msg) => Self::bad_request(msg),

101 ImportError::Unauthorized => Self::unauthorized(),

102 ImportError::Forbidden => Self::forbidden(),

103 ImportError::NotFound => Self::not_found("import job or source not found"),

104 ImportError::Conflict(msg) => Self::conflict(msg),

105 ImportError::RateLimited(msg) => Self::rate_limited(msg),

106 ImportError::Upstream(msg) => Self::build(

107 502,

108 export::rest::AppResponseData::err(502, "upstream forge request failed")

109 .with_error(msg)

110 .with_error_code("import_upstream_failed"),

111 ),

112 ImportError::Internal(message) => Self::internal(message),

113 }

114 }

115}

Added lib/import/src/identity.rs +220 −0 View lib/import/src/identity.rs

@@ -0,0 +1,220 @@

1use std::collections::HashMap;

2

3use models::git::git_commits;

4use models::users::{user, user_emails, user_external_identity};

5use sea_orm::{

6 ActiveModelTrait, ColumnTrait, DatabaseConnection, EntityTrait, QueryFilter, QuerySelect, Set,

7};

8use time::OffsetDateTime;

9use uuid::Uuid;

10

11use crate::errors::ImportError;

12use crate::providers::ImportProvider;

13use crate::service::ImportOptions;

14

15struct Candidate {

16 login: String,

17 external_id: Option<String>,

18 name: Option<String>,

19 email: Option<String>,

20 avatar_url: Option<String>,

21}

22

23pub fn login_from_noreply_email(email: &str) -> Option<String> {

24 let (local, domain) = email.split_once('@')?;

25 let domain = domain.to_ascii_lowercase();

26 let known = domain.ends_with("users.noreply.github.com")

27 || domain.ends_with("users.noreply.gitlab.com")

28 || domain.starts_with("noreply.");

29 if !known {

30 return None;

31 }

32 let login = match local.split_once('+') {

33 Some((_, rest)) if !rest.is_empty() => rest,

34 _ => local,

35 };

36 let login = login.trim();

37 if login.is_empty() {

38 None

39 } else {

40 Some(login.to_string())

41 }

42}

43

44pub async fn harvest(

45 db: &DatabaseConnection,

46 provider: ImportProvider,

47 repo_id: Uuid,

48 options: &ImportOptions,

49) -> Result<(), ImportError> {

50 let mut candidates: Vec<Candidate> = Vec::new();

51 if !options.owner.trim().is_empty() {

52 candidates.push(Candidate {

53 login: options.owner.clone(),

54 external_id: options.owner_external_id.clone(),

55 name: None,

56 email: None,

57 avatar_url: options.owner_avatar_url.clone(),

58 });

59 }

60

61 let authors: Vec<(String, String)> = git_commits::Entity::find()

62 .select_only()

63 .column(git_commits::Column::AuthorName)

64 .column(git_commits::Column::AuthorEmail)

65 .filter(git_commits::Column::RepoId.eq(repo_id))

66 .filter(git_commits::Column::AuthorId.is_null())

67 .distinct()

68 .into_tuple()

69 .all(db)

70 .await?;

71 let mut seen: Vec<String> = Vec::new();

72 for (name, email) in authors {

73 let Some(login) = login_from_noreply_email(&email) else {

74 continue;

75 };

76 let lower = login.to_lowercase();

77 if seen.contains(&lower) {

78 continue;

79 }

80 seen.push(lower);

81 candidates.push(Candidate {

82 login,

83 external_id: None,

84 name: Some(name),

85 email: Some(email),

86 avatar_url: None,

87 });

88 }

89

90 let mut resolved: HashMap<String, Uuid> = HashMap::new();

91 for candidate in candidates {

92 if let Some((email, uid)) = upsert(db, provider, &candidate).await? {

93 resolved.insert(email, uid);

94 }

95 }

96 backfill_commit_authors(db, repo_id, &resolved).await?;

97 Ok(())

98}

99

100async fn resolve_user(

101 db: &DatabaseConnection,

102 candidate: &Candidate,

103 lower_login: &str,

104) -> Result<Option<Uuid>, ImportError> {

105 if let Some(email) = candidate

106 .email

107 .as_deref()

108 .map(str::trim)

109 .filter(|v| !v.is_empty())

110 {

111 let row = user_emails::Entity::find()

112 .filter(user_emails::Column::LowerEmail.eq(email.to_lowercase()))

113 .one(db)

114 .await?;

115 if let Some(row) = row {

116 return Ok(Some(row.user_id));

117 }

118 }

119 let row = user::Entity::find()

120 .filter(user::Column::LowerName.eq(lower_login))

121 .one(db)

122 .await?;

123 Ok(row.map(|row| row.uid))

124}

125

126async fn upsert(

127 db: &DatabaseConnection,

128 provider: ImportProvider,

129 candidate: &Candidate,

130) -> Result<Option<(String, Uuid)>, ImportError> {

131 let login = candidate.login.trim();

132 if login.is_empty() {

133 return Ok(None);

134 }

135 let lower_login = login.to_lowercase();

136 let lower_email = candidate

137 .email

138 .as_deref()

139 .map(str::trim)

140 .filter(|v| !v.is_empty())

141 .map(str::to_lowercase);

142 let uid = resolve_user(db, candidate, &lower_login).await?;

143 let now = OffsetDateTime::now_utc();

144 let existing = user_external_identity::Entity::find()

145 .filter(user_external_identity::Column::Provider.eq(provider.as_str()))

146 .filter(user_external_identity::Column::LowerLogin.eq(&lower_login))

147 .one(db)

148 .await?;

149

150 let resolved_uid = match existing {

151 Some(row) => {

152 let mut model = user_external_identity::ActiveModel {

153 id: Set(row.id),

154 updated_at: Set(now),

155 ..Default::default()

156 };

157 if let Some(value) = candidate.external_id.clone().filter(|v| !v.is_empty()) {

158 model.external_id = Set(Some(value));

159 }

160 if let Some(value) = candidate.name.clone().filter(|v| !v.is_empty()) {

161 model.name = Set(Some(value));

162 }

163 if let Some(value) = candidate.avatar_url.clone().filter(|v| !v.is_empty()) {

164 model.avatar_url = Set(Some(value));

165 }

166 if let Some(value) = lower_email.clone() {

167 model.email = Set(Some(value));

168 }

169 let resolved = row.user_id.or(uid);

170 if row.user_id.is_none() && resolved.is_some() {

171 model.user_id = Set(resolved);

172 }

173 model.update(db).await?;

174 resolved

175 }

176 None => {

177 let row = user_external_identity::ActiveModel {

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

179 provider: Set(provider.as_str().to_string()),

180 external_id: Set(candidate.external_id.clone()),

181 lower_login: Set(lower_login),

182 login: Set(login.to_string()),

183 name: Set(candidate.name.clone()),

184 email: Set(lower_email.clone()),

185 avatar_url: Set(candidate.avatar_url.clone()),

186 user_id: Set(uid),

187 created_at: Set(now),

188 updated_at: Set(now),

189 }

190 .insert(db)

191 .await?;

192 row.user_id

193 }

194 };

195

196 Ok(match (candidate.email.clone(), resolved_uid, lower_email) {

197 (Some(email), Some(uid), Some(_)) if !email.trim().is_empty() => Some((email, uid)),

198 _ => None,

199 })

200}

201

202async fn backfill_commit_authors(

203 db: &DatabaseConnection,

204 repo_id: Uuid,

205 resolved: &HashMap<String, Uuid>,

206) -> Result<(), ImportError> {

207 for (email, uid) in resolved {

208 git_commits::Entity::update_many()

209 .col_expr(

210 git_commits::Column::AuthorId,

211 sea_orm::sea_query::Expr::value(*uid),

212 )

213 .filter(git_commits::Column::RepoId.eq(repo_id))

214 .filter(git_commits::Column::AuthorId.is_null())

215 .filter(git_commits::Column::AuthorEmail.eq(email.clone()))

216 .exec(db)

217 .await?;

218 }

219 Ok(())

220}

Added lib/import/src/lib.rs +17 −0 View lib/import/src/lib.rs

@@ -0,0 +1,17 @@

1pub mod errors;

2pub mod handlers;

3pub mod identity;

4pub mod openapi;

5pub mod providers;

6pub mod routes;

7pub mod service;

8pub mod share;

9pub mod worker;

10

11pub use errors::{ApiError, ImportError};

12pub use providers::{ImportProvider, ProbeInfo};

13pub use service::{

14 CreateImportCmd, ImportJobDto, ImportJobPage, ImportJobState, ImportProbeDto, ImportService,

15 ImportStage, ImportStatsDto, ListImportQuery, ProbeImportCmd, RequeueResultDto, RetryImportCmd,

16};

17pub use worker::ImportWorker;

Added lib/import/src/openapi.rs +39 −0 View lib/import/src/openapi.rs

@@ -0,0 +1,39 @@

1use utoipa::OpenApi;

2

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

4

5#[derive(OpenApi)]

6#[openapi(

7 info(title = "gitdata import api", version = "1.0"),

8 paths(

9 crate::handlers::probe_import::probe_import,

10 crate::handlers::create_import::create_import,

11 crate::handlers::list_imports::list_imports,

12 crate::handlers::get_import::get_import,

13 crate::handlers::cancel_import::cancel_import,

14 crate::handlers::retry_import::retry_import,

15 ),

16 components(schemas(

17 AppResponseData<crate::service::ImportProbeDto>,

18 AppResponseData<crate::service::ImportJobDto>,

19 AppResponseData<crate::service::ImportJobPage>,

20 NoData,

21 crate::service::ImportProbeDto,

22 crate::service::ImportJobDto,

23 crate::service::ImportJobPage,

24 crate::service::ImportStatsDto,

25 crate::service::ImportJobState,

26 crate::service::ImportStage,

27 crate::service::ImportSpacePath,

28 crate::service::ImportJobPath,

29 crate::service::ProbeImportCmd,

30 crate::service::CreateImportCmd,

31 crate::service::RetryImportCmd,

32 crate::service::ListImportQuery,

33 crate::providers::ImportProvider,

34 )),

35 tags(

36 (name = "imports", description = "GitHub / GitLab / Gitea 仓库导入迁移")

37 )

38)]

39pub struct ImportApiDoc;

Added lib/import/src/routes.rs +32 −0 View lib/import/src/routes.rs

@@ -0,0 +1,32 @@

1use axum::Router;

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

3

4use crate::handlers::cancel_import::cancel_import;

5use crate::handlers::create_import::create_import;

6use crate::handlers::get_import::get_import;

7use crate::handlers::list_imports::list_imports;

8use crate::handlers::probe_import::probe_import;

9use crate::handlers::requeue_imports::requeue_imports;

10use crate::handlers::retry_import::retry_import;

11use crate::service::ImportService;

12use crate::share::guard::import_guard;

13

14pub fn router(service: ImportService) -> Router<ImportService> {

15 Router::new()

16 .route(

17 "/imports/{space_name}",

18 post(create_import).get(list_imports),

19 )

20 .route("/imports/{space_name}/probe", post(probe_import))

21 .route("/imports/{space_name}/{job_id}", get(get_import))

22 .route("/imports/{space_name}/{job_id}/cancel", post(cancel_import))

23 .route("/imports/{space_name}/{job_id}/retry", post(retry_import))

24 .route_layer(axum::middleware::from_fn_with_state(

25 service.clone(),

26 import_guard,

27 ))

28}

29

30pub fn internal_router() -> Router<ImportService> {

31 Router::new().route("/imports/requeue", post(requeue_imports))

32}

Added lib/import/src/service.rs +690 −0 View lib/import/src/service.rs

@@ -0,0 +1,690 @@

1use config::AppConfig;

2use config::import::ImportConfig;

3use export::AppState;

4use export::pagination::Page;

5use git::service::GitService;

6use git::service::repos::CreateRepoCmd;

7use models::import::job;

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

9use sea_orm::{

10 ActiveModelTrait, ColumnTrait, DatabaseConnection, EntityTrait, PaginatorTrait, QueryFilter,

11 QueryOrder, QuerySelect, Set,

12};

13use serde::{Deserialize, Serialize};

14use time::OffsetDateTime;

15use utoipa::{IntoParams, ToSchema};

16use uuid::Uuid;

17

18use crate::errors::ImportError;

19use crate::providers::{self, ImportProvider, ProbeInfo};

20

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

22pub struct ImportOptions {

23 pub owner: String,

24 pub owner_external_id: Option<String>,

25 pub owner_avatar_url: Option<String>,

26 pub source_default_branch: Option<String>,

27 pub source_size_bytes: Option<i64>,

28 pub source_is_private: bool,

29 pub credential_username: Option<String>,

30 pub enable_lfs: bool,

31}

32

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

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

35pub enum ImportJobState {

36 Queued,

37 Processing,

38 Succeeded,

39 Failed,

40 Canceled,

41}

42

43impl ImportJobState {

44 #[must_use]

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

46 match self {

47 Self::Queued => "queued",

48 Self::Processing => "processing",

49 Self::Succeeded => "succeeded",

50 Self::Failed => "failed",

51 Self::Canceled => "canceled",

52 }

53 }

54

55 #[must_use]

56 pub fn parse(value: &str) -> Self {

57 match value {

58 "processing" => Self::Processing,

59 "succeeded" => Self::Succeeded,

60 "failed" => Self::Failed,

61 "canceled" => Self::Canceled,

62 _ => Self::Queued,

63 }

64 }

65

66 #[must_use]

67 pub fn is_terminal(self) -> bool {

68 matches!(self, Self::Succeeded | Self::Failed | Self::Canceled)

69 }

70}

71

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

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

74pub enum ImportStage {

75 Probe,

76 Clone,

77 Sync,

78 Finalize,

79 Done,

80}

81

82impl ImportStage {

83 #[must_use]

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

85 match self {

86 Self::Probe => "probe",

87 Self::Clone => "clone",

88 Self::Sync => "sync",

89 Self::Finalize => "finalize",

90 Self::Done => "done",

91 }

92 }

93

94 #[must_use]

95 pub fn parse(value: &str) -> Self {

96 match value {

97 "clone" => Self::Clone,

98 "sync" => Self::Sync,

99 "finalize" => Self::Finalize,

100 "done" => Self::Done,

101 _ => Self::Probe,

102 }

103 }

104}

105

106#[derive(Debug, Clone, Serialize, ToSchema)]

107pub struct ImportProbeDto {

108 pub provider: ImportProvider,

109 pub owner: String,

110 pub name: String,

111 pub clone_url: String,

112 pub default_branch: String,

113 pub is_private: bool,

114 pub description: Option<String>,

115 pub size_bytes: Option<i64>,

116}

117

118impl From<&ProbeInfo> for ImportProbeDto {

119 fn from(info: &ProbeInfo) -> Self {

120 Self {

121 provider: info.provider,

122 owner: info.owner.clone(),

123 name: info.name.clone(),

124 clone_url: info.clone_url.clone(),

125 default_branch: info.default_branch.clone(),

126 is_private: info.is_private,

127 description: info.description.clone(),

128 size_bytes: info.size_bytes,

129 }

130 }

131}

132

133#[derive(Debug, Clone, Default, Serialize, Deserialize, ToSchema, PartialEq)]

134pub struct ImportStatsDto {

135 pub branches: i64,

136 pub tags: i64,

137 pub commits: i64,

138 pub size_bytes: i64,

139 pub default_branch: Option<String>,

140 pub warnings: Vec<String>,

141}

142

143#[derive(Debug, Clone, Serialize, ToSchema)]

144pub struct ImportJobDto {

145 pub id: Uuid,

146 pub provider: ImportProvider,

147 pub source_url: String,

148 pub base_url: Option<String>,

149 pub clone_url: String,

150 pub space_id: Uuid,

151 pub space_name: String,

152 pub repo_id: Option<Uuid>,

153 pub repo_name: String,

154 pub state: ImportJobState,

155 pub stage: ImportStage,

156 pub attempts: i32,

157 pub last_error: Option<String>,

158 pub stats: ImportStatsDto,

159 pub created_by: Uuid,

160 pub created_at: OffsetDateTime,

161 pub updated_at: OffsetDateTime,

162 pub started_at: Option<OffsetDateTime>,

163 pub finished_at: Option<OffsetDateTime>,

164}

165

166pub type ImportJobPage = Page<ImportJobDto>;

167

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

169pub struct ProbeImportCmd {

170 pub provider: ImportProvider,

171 pub source: String,

172 #[serde(default)]

173 pub base_url: Option<String>,

174 #[serde(default)]

175 pub token: Option<String>,

176}

177

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

179pub struct CreateImportCmd {

180 pub provider: ImportProvider,

181 pub source: String,

182 #[serde(default)]

183 pub base_url: Option<String>,

184 #[serde(default)]

185 pub token: Option<String>,

186 #[serde(default)]

187 pub name: Option<String>,

188 #[serde(default)]

189 pub description: Option<String>,

190 #[serde(default)]

191 pub is_private: Option<bool>,

192 #[serde(default)]

193 pub parent_dir_id: Option<Uuid>,

194 #[serde(default)]

195 pub enable_lfs: bool,

196}

197

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

199pub struct RetryImportCmd {

200 #[serde(default)]

201 pub token: Option<String>,

202}

203

204#[derive(Debug, Clone, Deserialize, IntoParams, ToSchema)]

205pub struct ListImportQuery {

206 pub state: Option<String>,

207 pub repo_name: Option<String>,

208 pub limit: Option<u64>,

209 pub offset: Option<u64>,

210}

211

212#[derive(Debug, Clone, Serialize)]

213pub struct RequeueResultDto {

214 pub failed: u64,

215 pub stale: u64,

216}

217

218#[derive(Debug, Clone, Deserialize, IntoParams, ToSchema)]

219pub struct ImportSpacePath {

220 pub space_name: String,

221}

222

223#[derive(Debug, Clone, Deserialize, IntoParams, ToSchema)]

224pub struct ImportJobPath {

225 pub space_name: String,

226 pub job_id: Uuid,

227}

228

229#[derive(Clone)]

230pub struct ImportService {

231 pub state: AppState,

232 pub git: GitService,

233}

234

235impl ImportService {

236 pub fn new(state: AppState) -> Self {

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

238 Self { state, git }

239 }

240

241 pub fn config(&self) -> &AppConfig {

242 &self.state.config

243 }

244

245 pub fn import_config(&self) -> &ImportConfig {

246 &self.state.config.import

247 }

248

249 pub fn db_read(&self) -> &DatabaseConnection {

250 self.state.database.read()

251 }

252

253 pub fn db_write(&self) -> &DatabaseConnection {

254 self.state.database.write()

255 }

256

257 fn is_platform_admin(&self, caller: Uuid) -> bool {

258 self.config().apihost.is_platform_admin(&caller.to_string())

259 }

260

261 pub fn dto(&self, row: &job::Model, space_name: &str) -> ImportJobDto {

262 ImportJobDto {

263 id: row.id,

264 provider: ImportProvider::parse(&row.provider).unwrap_or(ImportProvider::Github),

265 source_url: row.source_url.clone(),

266 base_url: row.base_url.clone(),

267 clone_url: row.clone_url.clone(),

268 space_id: row.space_id,

269 space_name: space_name.to_string(),

270 repo_id: row.repo_id,

271 repo_name: row.repo_name.clone(),

272 state: ImportJobState::parse(&row.state),

273 stage: ImportStage::parse(&row.stage),

274 attempts: row.attempts,

275 last_error: row.last_error.clone(),

276 stats: serde_json::from_value(row.stats.clone()).unwrap_or_default(),

277 created_by: row.created_by,

278 created_at: row.created_at,

279 updated_at: row.updated_at,

280 started_at: row.started_at,

281 finished_at: row.finished_at,

282 }

283 }

284

285 pub async fn lookup_space(&self, space_name: &str) -> Result<spaces::Model, ImportError> {

286 let lower = space_name.trim().to_lowercase();

287 if lower.is_empty() {

288 return Err(ImportError::NotFound);

289 }

290 spaces::Entity::find()

291 .filter(spaces::Column::LowerName.eq(&lower))

292 .filter(spaces::Column::IsActive.eq(true))

293 .one(self.db_read())

294 .await?

295 .ok_or(ImportError::NotFound)

296 }

297

298 async fn require_member(&self, space: &spaces::Model, caller: Uuid) -> Result<(), ImportError> {

299 if space.owner_id == caller || self.is_platform_admin(caller) {

300 return Ok(());

301 }

302 let member = space_members::Entity::find_by_id((space.id, caller))

303 .one(self.db_read())

304 .await?;

305 let active = member

306 .is_some_and(|row| row.status == models::space::space_members::MemberStatus::Active);

307 if active {

308 Ok(())

309 } else {

310 Err(ImportError::Forbidden)

311 }

312 }

313

314 pub async fn probe_source(

315 &self,

316 provider: ImportProvider,

317 source: &str,

318 base_url: Option<&str>,

319 token: Option<&str>,

320 ) -> Result<ProbeInfo, ImportError> {

321 let config = self.import_config();

322 if !config.enabled {

323 return Err(ImportError::Disabled);

324 }

325 if let Some(value) = base_url.map(str::trim).filter(|value| !value.is_empty()) {

326 providers::ensure_url_allowed(value, config)?;

327 }

328 providers::probe(providers::ProbeRequest {

329 provider,

330 source,

331 base_url,

332 token,

333 config,

334 })

335 .await

336 }

337

338 pub async fn probe(

339 &self,

340 caller: Uuid,

341 space_name: &str,

342 cmd: ProbeImportCmd,

343 ) -> Result<ImportProbeDto, ImportError> {

344 let space = self.lookup_space(space_name).await?;

345 self.require_member(&space, caller).await?;

346 let info = self

347 .probe_source(

348 cmd.provider,

349 &cmd.source,

350 cmd.base_url.as_deref(),

351 cmd.token.as_deref(),

352 )

353 .await?;

354 providers::ensure_url_allowed(&info.clone_url, self.import_config())?;

355 Ok(ImportProbeDto::from(&info))

356 }

357

358 pub async fn create(

359 &self,

360 caller: Uuid,

361 space_name: &str,

362 cmd: CreateImportCmd,

363 ) -> Result<ImportJobDto, ImportError> {

364 let config = self.import_config().clone();

365 if !config.enabled {

366 return Err(ImportError::Disabled);

367 }

368 let space = self.lookup_space(space_name).await?;

369 if space.owner_id != caller && !self.is_platform_admin(caller) {

370 return Err(ImportError::Forbidden);

371 }

372 let info = self

373 .probe_source(

374 cmd.provider,

375 &cmd.source,

376 cmd.base_url.as_deref(),

377 cmd.token.as_deref(),

378 )

379 .await?;

380 providers::ensure_url_allowed(&info.clone_url, &config)?;

381 if config.max_repo_bytes > 0

382 && info.size_bytes.unwrap_or(0).max(0) as u64 > config.max_repo_bytes

383 {

384 return Err(ImportError::BadRequest(format!(

385 "source repository exceeds the {} byte import limit",

386 config.max_repo_bytes

387 )));

388 }

389

390 let repo_name = match cmd.name.as_deref().map(str::trim).filter(|v| !v.is_empty()) {

391 Some(name) => name.to_string(),

392 None => info.name.clone(),

393 };

394 let description = cmd

395 .description

396 .clone()

397 .filter(|value| !value.trim().is_empty())

398 .or_else(|| info.description.clone());

399 let is_private = cmd.is_private.unwrap_or(info.is_private);

400 let init_message = format!("Import {} from {}", repo_name, info.clone_url);

401 let repo = self

402 .git

403 .create_repo(

404 space.name.clone(),

405 caller,

406 CreateRepoCmd {

407 name: repo_name.clone(),

408 description,

409 is_private,

410 default_branch: Some(info.default_branch.clone()),

411 parent_dir_id: cmd.parent_dir_id,

412 enable_lfs: cmd.enable_lfs,

413 license_name: None,

414 init_message: Some(init_message),

415 },

416 )

417 .await?;

418

419 let token_enc = match cmd

420 .token

421 .as_deref()

422 .map(str::trim)

423 .filter(|v| !v.is_empty())

424 {

425 Some(token) => Some(encrypt_token(&config.token_secret, token)?),

426 None => None,

427 };

428 let options = ImportOptions {

429 owner: info.owner.clone(),

430 owner_external_id: info.owner_external_id.clone(),

431 owner_avatar_url: info.owner_avatar_url.clone(),

432 source_default_branch: Some(info.default_branch.clone()),

433 source_size_bytes: info.size_bytes,

434 source_is_private: info.is_private,

435 credential_username: info.credential_username.clone(),

436 enable_lfs: cmd.enable_lfs,

437 };

438 let now = OffsetDateTime::now_utc();

439 let row = job::ActiveModel {

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

441 provider: Set(info.provider.as_str().to_string()),

442 source_url: Set(cmd.source.trim().to_string()),

443 base_url: Set(cmd

444 .base_url

445 .as_deref()

446 .map(str::trim)

447 .filter(|v| !v.is_empty())

448 .map(str::to_string)),

449 clone_url: Set(info.clone_url.clone()),

450 space_id: Set(space.id),

451 repo_id: Set(Some(repo.id)),

452 repo_name: Set(repo_name),

453 created_by: Set(caller),

454 state: Set(ImportJobState::Queued.as_str().to_string()),

455 stage: Set(ImportStage::Clone.as_str().to_string()),

456 attempts: Set(0),

457 last_error: Set(None),

458 token_enc: Set(token_enc),

459 options: Set(serde_json::to_value(&options).unwrap_or_default()),

460 stats: Set(serde_json::to_value(ImportStatsDto::default()).unwrap_or_default()),

461 run_after: Set(now),

462 locked_at: Set(None),

463 started_at: Set(None),

464 finished_at: Set(None),

465 created_at: Set(now),

466 updated_at: Set(now),

467 }

468 .insert(self.db_write())

469 .await?;

470 Ok(self.dto(&row, &space.name))

471 }

472

473 pub async fn list(

474 &self,

475 caller: Uuid,

476 space_name: &str,

477 query: ListImportQuery,

478 ) -> Result<ImportJobPage, ImportError> {

479 let space = self.lookup_space(space_name).await?;

480 self.require_member(&space, caller).await?;

481 let limit = query.limit.unwrap_or(20).clamp(1, 100);

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

483 let mut select = job::Entity::find().filter(job::Column::SpaceId.eq(space.id));

484 if space.owner_id != caller && !self.is_platform_admin(caller) {

485 select = select.filter(job::Column::CreatedBy.eq(caller));

486 }

487 if let Some(state) = query

488 .state

489 .as_deref()

490 .map(str::trim)

491 .filter(|v| !v.is_empty())

492 {

493 select = select.filter(job::Column::State.eq(state.to_string()));

494 }

495 if let Some(repo) = query

496 .repo_name

497 .as_deref()

498 .map(str::trim)

499 .filter(|v| !v.is_empty())

500 {

501 select = select.filter(job::Column::RepoName.eq(repo.to_string()));

502 }

503 let total = select.clone().count(self.db_read()).await? as i64;

504 let rows = select

505 .order_by_desc(job::Column::CreatedAt)

506 .limit(limit)

507 .offset(offset)

508 .all(self.db_read())

509 .await?;

510 let items = rows.iter().map(|row| self.dto(row, &space.name)).collect();

511 Ok(Page {

512 items,

513 total,

514 limit,

515 offset,

516 })

517 }

518

519 pub async fn load_job(

520 &self,

521 caller: Uuid,

522 space_name: &str,

523 job_id: Uuid,

524 ) -> Result<(spaces::Model, job::Model), ImportError> {

525 let space = self.lookup_space(space_name).await?;

526 self.require_member(&space, caller).await?;

527 let row = job::Entity::find_by_id(job_id)

528 .filter(job::Column::SpaceId.eq(space.id))

529 .one(self.db_read())

530 .await?

531 .ok_or(ImportError::NotFound)?;

532 if row.created_by != caller && space.owner_id != caller && !self.is_platform_admin(caller) {

533 return Err(ImportError::Forbidden);

534 }

535 Ok((space, row))

536 }

537

538 pub async fn get(

539 &self,

540 caller: Uuid,

541 space_name: &str,

542 job_id: Uuid,

543 ) -> Result<ImportJobDto, ImportError> {

544 let (space, row) = self.load_job(caller, space_name, job_id).await?;

545 Ok(self.dto(&row, &space.name))

546 }

547

548 pub async fn requeue_pending(&self) -> Result<RequeueResultDto, ImportError> {

549 let now = OffsetDateTime::now_utc();

550 let stale_cutoff = now

551 - time::Duration::seconds(

552 i64::try_from(self.import_config().stale_lock_secs.max(1)).unwrap_or(1800),

553 );

554 let failed = job::Entity::update_many()

555 .col_expr(

556 job::Column::State,

557 sea_orm::sea_query::Expr::value(ImportJobState::Queued.as_str()),

558 )

559 .col_expr(

560 job::Column::Stage,

561 sea_orm::sea_query::Expr::value(ImportStage::Clone.as_str()),

562 )

563 .col_expr(job::Column::Attempts, sea_orm::sea_query::Expr::value(0i32))

564 .col_expr(

565 job::Column::LockedAt,

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

567 )

568 .col_expr(

569 job::Column::FinishedAt,

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

571 )

572 .col_expr(job::Column::RunAfter, sea_orm::sea_query::Expr::value(now))

573 .col_expr(job::Column::UpdatedAt, sea_orm::sea_query::Expr::value(now))

574 .filter(job::Column::State.eq(ImportJobState::Failed.as_str()))

575 .exec(self.db_write())

576 .await?;

577 let stale = job::Entity::update_many()

578 .col_expr(

579 job::Column::State,

580 sea_orm::sea_query::Expr::value(ImportJobState::Queued.as_str()),

581 )

582 .col_expr(

583 job::Column::LockedAt,

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

585 )

586 .col_expr(job::Column::RunAfter, sea_orm::sea_query::Expr::value(now))

587 .col_expr(job::Column::UpdatedAt, sea_orm::sea_query::Expr::value(now))

588 .filter(job::Column::State.eq(ImportJobState::Processing.as_str()))

589 .filter(job::Column::LockedAt.lt(stale_cutoff))

590 .exec(self.db_write())

591 .await?;

592 Ok(RequeueResultDto {

593 failed: failed.rows_affected,

594 stale: stale.rows_affected,

595 })

596 }

597

598 pub async fn cancel(

599 &self,

600 caller: Uuid,

601 space_name: &str,

602 job_id: Uuid,

603 ) -> Result<ImportJobDto, ImportError> {

604 let (space, row) = self.load_job(caller, space_name, job_id).await?;

605 let state = ImportJobState::parse(&row.state);

606 if state.is_terminal() {

607 return Err(ImportError::Conflict(format!(

608 "import job is already {}",

609 state.as_str()

610 )));

611 }

612 let now = OffsetDateTime::now_utc();

613 let updated = job::ActiveModel {

614 id: Set(row.id),

615 state: Set(ImportJobState::Canceled.as_str().to_string()),

616 token_enc: Set(None),

617 locked_at: Set(None),

618 finished_at: Set(Some(now)),

619 updated_at: Set(now),

620 ..Default::default()

621 }

622 .update(self.db_write())

623 .await?;

624 Ok(self.dto(&updated, &space.name))

625 }

626

627 pub async fn retry(

628 &self,

629 caller: Uuid,

630 space_name: &str,

631 job_id: Uuid,

632 cmd: RetryImportCmd,

633 ) -> Result<ImportJobDto, ImportError> {

634 let (space, row) = self.load_job(caller, space_name, job_id).await?;

635 let state = ImportJobState::parse(&row.state);

636 if !matches!(state, ImportJobState::Failed | ImportJobState::Canceled) {

637 return Err(ImportError::Conflict(format!(

638 "only failed or canceled import jobs can be retried, current state is {}",

639 state.as_str()

640 )));

641 }

642 let config = self.import_config().clone();

643 let token_enc = match cmd

644 .token

645 .as_deref()

646 .map(str::trim)

647 .filter(|v| !v.is_empty())

648 {

649 Some(token) => Some(encrypt_token(&config.token_secret, token)?),

650 None => None,

651 };

652 let now = OffsetDateTime::now_utc();

653 let updated = job::ActiveModel {

654 id: Set(row.id),

655 state: Set(ImportJobState::Queued.as_str().to_string()),

656 stage: Set(ImportStage::Clone.as_str().to_string()),

657 attempts: Set(0),

658 last_error: Set(None),

659 token_enc: Set(token_enc),

660 run_after: Set(now),

661 locked_at: Set(None),

662 finished_at: Set(None),

663 updated_at: Set(now),

664 ..Default::default()

665 }

666 .update(self.db_write())

667 .await?;

668 Ok(self.dto(&updated, &space.name))

669 }

670}

671

672pub fn encrypt_token(secret: &str, token: &str) -> Result<String, ImportError> {

673 auth::share::crypto::aes_encrypt(secret, token)

674 .map_err(|e| ImportError::Internal(format!("token encryption failed: {e}")))

675}

676

677pub fn decrypt_token(secret: &str, stored: &str) -> Result<String, ImportError> {

678 auth::share::crypto::aes_decrypt(secret, stored)

679 .map_err(|e| ImportError::Internal(format!("token decryption failed: {e}")))

680}

681

682#[async_trait::async_trait]

683impl session::SessionGuard for ImportService {

684 async fn lookup_session_with_user(

685 &self,

686 token_hash: &str,

687 ) -> Result<Option<session::SessionWithUser>, session::SessionError> {

688 session::lookup_session_with_user(self.db_read(), token_hash).await

689 }

690}

Added lib/import/src/worker.rs +424 −0 View lib/import/src/worker.rs

@@ -0,0 +1,424 @@

1use std::path::{Path, PathBuf};

2use std::sync::Arc;

3use std::time::Duration;

4

5use cache::CacheClient;

6use config::git::SyncSettings;

7use config::import::ImportConfig;

8use export::{AppDatabase, AppRedis};

9use git::service::clone::{GitCredential, mirror_fetch};

10use git::sync::RepoSync;

11use models::import::job;

12use models::repo::repos;

13use sea_orm::sea_query::Expr;

14use sea_orm::{ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, QueryOrder, Set};

15use time::OffsetDateTime;

16use tokio::sync::Semaphore;

17use tokio_util::sync::CancellationToken;

18use tracing::{info, warn};

19use uuid::Uuid;

20

21use crate::errors::ImportError;

22use crate::identity;

23use crate::providers::ImportProvider;

24use crate::service::{ImportJobState, ImportOptions, ImportStage, ImportStatsDto, decrypt_token};

25

26#[derive(Clone)]

27pub struct ImportWorker {

28 db: Arc<AppDatabase>,

29 redis: AppRedis,

30 cache: Arc<CacheClient>,

31 repo_root: PathBuf,

32 settings: SyncSettings,

33 config: ImportConfig,

34 cancel: CancellationToken,

35}

36

37impl ImportWorker {

38 #[must_use]

39 pub fn new(

40 db: Arc<AppDatabase>,

41 redis: AppRedis,

42 cache: Arc<CacheClient>,

43 repo_root: PathBuf,

44 settings: SyncSettings,

45 config: ImportConfig,

46 cancel: CancellationToken,

47 ) -> Self {

48 Self {

49 db,

50 redis,

51 cache,

52 repo_root,

53 settings,

54 config,

55 cancel,

56 }

57 }

58

59 pub async fn run(self) {

60 let permits = Arc::new(Semaphore::new(self.config.max_concurrent_jobs.max(1)));

61 let delay = Duration::from_secs(self.config.poll_interval_secs.max(1));

62 info!(

63 concurrency = self.config.max_concurrent_jobs.max(1),

64 "import worker started"

65 );

66 loop {

67 tokio::select! {

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

69 _ = tokio::time::sleep(delay) => {

70 if let Err(error) = self.reap_stale().await {

71 warn!(%error, "import stale job reap failed");

72 }

73 loop {

74 let Ok(permit) = permits.clone().try_acquire_owned() else {

75 break;

76 };

77 match self.claim().await {

78 Ok(Some(row)) => {

79 let worker = self.clone();

80 tokio::spawn(async move {

81 let _permit = permit;

82 worker.process(row).await;

83 });

84 }

85 Ok(None) => {

86 drop(permit);

87 break;

88 }

89 Err(error) => {

90 drop(permit);

91 warn!(%error, "import job claim failed");

92 break;

93 }

94 }

95 }

96 }

97 }

98 }

99 info!("import worker stopped");

100 }

101

102 async fn claim(&self) -> Result<Option<job::Model>, ImportError> {

103 let now = OffsetDateTime::now_utc();

104 let Some(row) = job::Entity::find()

105 .filter(job::Column::State.eq(ImportJobState::Queued.as_str()))

106 .filter(job::Column::RunAfter.lte(now))

107 .order_by_asc(job::Column::CreatedAt)

108 .one(self.db.read())

109 .await?

110 else {

111 return Ok(None);

112 };

113 let claimed = job::Entity::update_many()

114 .col_expr(

115 job::Column::State,

116 Expr::value(ImportJobState::Processing.as_str()),

117 )

118 .col_expr(job::Column::LockedAt, Expr::value(now))

119 .col_expr(job::Column::StartedAt, Expr::value(now))

120 .col_expr(job::Column::UpdatedAt, Expr::value(now))

121 .filter(job::Column::Id.eq(row.id))

122 .filter(job::Column::State.eq(ImportJobState::Queued.as_str()))

123 .exec(self.db.write())

124 .await?;

125 if claimed.rows_affected == 0 {

126 return Ok(None);

127 }

128 Ok(job::Entity::find_by_id(row.id).one(self.db.read()).await?)

129 }

130

131 async fn reap_stale(&self) -> Result<(), ImportError> {

132 let cutoff = OffsetDateTime::now_utc()

133 - time::Duration::seconds(

134 i64::try_from(self.config.stale_lock_secs.max(1)).unwrap_or(1800),

135 );

136 let now = OffsetDateTime::now_utc();

137 let result = job::Entity::update_many()

138 .col_expr(

139 job::Column::State,

140 Expr::value(ImportJobState::Queued.as_str()),

141 )

142 .col_expr(

143 job::Column::LockedAt,

144 Expr::value(Option::<OffsetDateTime>::None),

145 )

146 .col_expr(job::Column::RunAfter, Expr::value(now))

147 .col_expr(job::Column::UpdatedAt, Expr::value(now))

148 .filter(job::Column::State.eq(ImportJobState::Processing.as_str()))

149 .filter(job::Column::LockedAt.lt(cutoff))

150 .exec(self.db.write())

151 .await?;

152 if result.rows_affected > 0 {

153 warn!(count = result.rows_affected, "requeued stale import jobs");

154 }

155 Ok(())

156 }

157

158 fn spawn_heartbeat(&self, job_id: Uuid) -> tokio::task::JoinHandle<()> {

159 let db = self.db.clone();

160 let cancel = self.cancel.clone();

161 let interval = Duration::from_secs((self.config.stale_lock_secs.max(3) / 3).max(1));

162 tokio::spawn(async move {

163 loop {

164 tokio::select! {

165 _ = cancel.cancelled() => break,

166 _ = tokio::time::sleep(interval) => {

167 let now = OffsetDateTime::now_utc();

168 let result = job::Entity::update_many()

169 .col_expr(job::Column::LockedAt, Expr::value(now))

170 .col_expr(job::Column::UpdatedAt, Expr::value(now))

171 .filter(job::Column::Id.eq(job_id))

172 .filter(job::Column::State.eq(ImportJobState::Processing.as_str()))

173 .exec(db.write())

174 .await;

175 if let Err(error) = result {

176 warn!(%error, "import heartbeat failed");

177 }

178 }

179 }

180 }

181 })

182 }

183

184 async fn process(&self, row: job::Model) {

185 let heartbeat = self.spawn_heartbeat(row.id);

186 let outcome = self.run_job(&row).await;

187 heartbeat.abort();

188 let stage = ImportStage::parse(&row.stage);

189 match outcome {

190 Ok(stats) => {

191 if let Err(error) = self

192 .finish(

193 row.id,

194 ImportJobState::Succeeded,

195 ImportStage::Done,

196 None,

197 Some(stats),

198 )

199 .await

200 {

201 warn!(%error, job_id = %row.id, "failed to mark import job succeeded");

202 return;

203 }

204 info!(job_id = %row.id, repo = %row.repo_name, "repository import completed");

205 }

206 Err(ImportError::Canceled) => {

207 let _ = self

208 .finish(row.id, ImportJobState::Canceled, stage, None, None)

209 .await;

210 info!(job_id = %row.id, "repository import canceled");

211 }

212 Err(error) => {

213 let attempts = row.attempts + 1;

214 let exhausted = attempts >= self.config.max_attempts.max(1);

215 if error.is_permanent() || exhausted {

216 let _ = self

217 .finish(

218 row.id,

219 ImportJobState::Failed,

220 stage,

221 Some(error.to_string()),

222 None,

223 )

224 .await;

225 warn!(job_id = %row.id, attempts, %error, "repository import failed");

226 } else {

227 let _ = self.requeue(row.id, attempts, error.to_string()).await;

228 warn!(job_id = %row.id, attempts, %error, "repository import retried later");

229 }

230 }

231 }

232 }

233

234 async fn set_stage(&self, id: Uuid, stage: ImportStage) -> Result<(), ImportError> {

235 let now = OffsetDateTime::now_utc();

236 job::ActiveModel {

237 id: Set(id),

238 stage: Set(stage.as_str().to_string()),

239 updated_at: Set(now),

240 ..Default::default()

241 }

242 .update(self.db.write())

243 .await?;

244 Ok(())

245 }

246

247 async fn finish(

248 &self,

249 id: Uuid,

250 state: ImportJobState,

251 stage: ImportStage,

252 last_error: Option<String>,

253 stats: Option<ImportStatsDto>,

254 ) -> Result<(), ImportError> {

255 let now = OffsetDateTime::now_utc();

256 let mut model = job::ActiveModel {

257 id: Set(id),

258 state: Set(state.as_str().to_string()),

259 stage: Set(stage.as_str().to_string()),

260 last_error: Set(last_error),

261 token_enc: Set(None),

262 locked_at: Set(None),

263 finished_at: Set(Some(now)),

264 updated_at: Set(now),

265 ..Default::default()

266 };

267 if let Some(stats) = stats {

268 model.stats = Set(serde_json::to_value(&stats).unwrap_or_default());

269 }

270 model.update(self.db.write()).await?;

271 Ok(())

272 }

273

274 async fn requeue(&self, id: Uuid, attempts: i32, error: String) -> Result<(), ImportError> {

275 let now = OffsetDateTime::now_utc();

276 let backoff = time::Duration::seconds(i64::from(attempts.clamp(1, 8)) * 15);

277 job::ActiveModel {

278 id: Set(id),

279 state: Set(ImportJobState::Queued.as_str().to_string()),

280 attempts: Set(attempts),

281 last_error: Set(Some(error)),

282 locked_at: Set(None),

283 run_after: Set(now + backoff),

284 updated_at: Set(now),

285 ..Default::default()

286 }

287 .update(self.db.write())

288 .await?;

289 Ok(())

290 }

291

292 async fn check_canceled(&self, id: Uuid) -> Result<(), ImportError> {

293 let row = job::Entity::find_by_id(id).one(self.db.read()).await?;

294 match row {

295 None => Err(ImportError::Internal(

296 "import job disappeared while processing".to_string(),

297 )),

298 Some(row) if ImportJobState::parse(&row.state) == ImportJobState::Canceled => {

299 Err(ImportError::Canceled)

300 }

301 Some(_) => Ok(()),

302 }

303 }

304

305 async fn run_job(&self, row: &job::Model) -> Result<ImportStatsDto, ImportError> {

306 let repo_id = row.repo_id.ok_or_else(|| {

307 ImportError::Internal("import job is not attached to a repository".to_string())

308 })?;

309 let repo = repos::Entity::find_by_id(repo_id)

310 .one(self.db.read())

311 .await?

312 .ok_or_else(|| {

313 ImportError::Internal(format!("repository {repo_id} no longer exists"))

314 })?;

315 let provider = ImportProvider::parse(&row.provider)?;

316 let options: ImportOptions =

317 serde_json::from_value(row.options.clone()).unwrap_or_default();

318

319 self.check_canceled(row.id).await?;

320 self.set_stage(row.id, ImportStage::Clone).await?;

321 let credential = match row.token_enc.as_deref() {

322 Some(stored) => {

323 let token = decrypt_token(&self.config.token_secret, stored)?;

324 let username = options

325 .credential_username

326 .clone()

327 .unwrap_or_else(|| provider.default_credential_username().to_string());

328 Some(GitCredential::new(username, token))

329 }

330 None => None,

331 };

332 let dir = git::context::facade::repo_dir_for(&self.repo_root, repo_id);

333 let result = mirror_fetch(

334 "import::mirror",

335 &dir,

336 &row.clone_url,

337 credential.as_ref(),

338 self.config.clone_timeout_secs.max(1),

339 )

340 .await;

341 drop(credential);

342 result?;

343

344 self.check_canceled(row.id).await?;

345 self.set_stage(row.id, ImportStage::Sync).await?;

346 let sync = RepoSync::new(

347 self.db.clone(),

348 self.redis.clone(),

349 self.cache.clone(),

350 repo,

351 self.repo_root.clone(),

352 self.settings.clone(),

353 );

354 sync.sync()

355 .await

356 .map_err(|e| ImportError::Internal(format!("repository sync failed: {e}")))?;

357

358 self.check_canceled(row.id).await?;

359 self.set_stage(row.id, ImportStage::Finalize).await?;

360 let fresh = repos::Entity::find_by_id(repo_id)

361 .one(self.db.read())

362 .await?

363 .ok_or_else(|| {

364 ImportError::Internal(format!("repository {repo_id} disappeared during import"))

365 })?;

366 let default_branch = fresh

367 .default_branch

368 .strip_prefix("refs/heads/")

369 .unwrap_or(&fresh.default_branch)

370 .to_string();

371 let mut stats = ImportStatsDto {

372 branches: i64::from(fresh.nums_branches),

373 tags: i64::from(fresh.nums_tags),

374 commits: i64::from(fresh.nums_commits),

375 size_bytes: fresh.git_size,

376 default_branch: Some(default_branch),

377 warnings: Vec::new(),

378 };

379 if uses_lfs(

380 &dir,

381 &fresh.default_branch,

382 self.settings.git_command_timeout_secs,

383 )

384 .await

385 {

386 stats.warnings.push(

387 "源仓库使用 Git LFS，本次导入只搬运了指针文件，未搬迁 LFS 大对象。".to_string(),

388 );

389 }

390 if let Err(error) = identity::harvest(self.db.write(), provider, repo_id, &options).await {

391 warn!(%error, job_id = %row.id, "import identity harvest failed");

392 }

393 Ok(stats)

394 }

395}

396

397async fn uses_lfs(dir: &Path, default_branch: &str, timeout_secs: u64) -> bool {

398 let revision = default_branch

399 .strip_prefix("refs/heads/")

400 .unwrap_or(default_branch);

401 if revision.is_empty() {

402 return false;

403 }

404 let output = git::service::gitcmd::run_git_soft(

405 "import::lfs_probe",

406 dir,

407 vec![

408 "grep".to_string(),

409 "-l".to_string(),

410 "--fixed-strings".to_string(),

411 "filter=lfs".to_string(),

412 revision.to_string(),

413 "--".to_string(),

414 ".gitattributes".to_string(),

415 ],

416 &[],

417 timeout_secs.clamp(5, 60),

418 )

419 .await;

420 match output {

421 Ok(output) => output.status.success() && !output.stdout.is_empty(),

422 Err(_) => false,

423 }

424}

Added lib/import/Cargo.toml +26 −0 View lib/import/Cargo.toml

@@ -0,0 +1,26 @@

1[package]

2name = "import"

3version.workspace = true

4edition.workspace = true

5

6[dependencies]

7async-trait = { workspace = true }

8auth = { workspace = true }

9axum = { workspace = true }

10cache = { workspace = true }

11config = { workspace = true }

12export = { workspace = true }

13git = { workspace = true }

14models = { workspace = true }

15reqwest = { workspace = true }

16sea-orm = { workspace = true }

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

18serde_json = { workspace = true }

19session = { workspace = true }

20thiserror = { workspace = true }

21time = { workspace = true }

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

23tokio-util = { workspace = true }

24tracing = { workspace = true }

25utoipa = { workspace = true }

26uuid = { workspace = true }

No newline at end of file

Added lib/models/src/import/job.rs +49 −0 View lib/models/src/import/job.rs

@@ -0,0 +1,49 @@

1use sea_orm::JsonValue;

2use sea_orm::entity::prelude::*;

3use time::OffsetDateTime;

4use uuid::Uuid;

5

6#[sea_orm::model]

7#[derive(Clone, Debug, PartialEq, DeriveEntityModel)]

8#[sea_orm(table_name = "import_job")]

9pub struct Model {

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

11 pub id: Uuid,

12 #[sea_orm(indexed)]

13 pub provider: String,

14 pub source_url: String,

15 pub base_url: Option<String>,

16 pub clone_url: String,

17 #[sea_orm(indexed)]

18 pub space_id: Uuid,

19 #[sea_orm(indexed)]

20 pub repo_id: Option<Uuid>,

21 pub repo_name: String,

22 #[sea_orm(indexed)]

23 pub created_by: Uuid,

24 #[sea_orm(indexed)]

25 pub state: String,

26 pub stage: String,

27 #[sea_orm(default_value = 0)]

28 pub attempts: i32,

29 pub last_error: Option<String>,

30 pub token_enc: Option<String>,

31 #[sea_orm(column_type = "JsonBinary")]

32 pub options: JsonValue,

33 #[sea_orm(column_type = "JsonBinary")]

34 pub stats: JsonValue,

35 pub run_after: OffsetDateTime,

36 pub locked_at: Option<OffsetDateTime>,

37 pub started_at: Option<OffsetDateTime>,

38 pub finished_at: Option<OffsetDateTime>,

39 pub created_at: OffsetDateTime,

40 pub updated_at: OffsetDateTime,

41 #[sea_orm(belongs_to, from = "space_id", to = "id")]

42 pub space: BelongsTo<super::super::space::spaces::Entity>,

43 #[sea_orm(belongs_to, from = "repo_id", to = "id")]

44 pub repo: BelongsTo<Option<super::super::repo::repos::Entity>>,

45 #[sea_orm(belongs_to, from = "created_by", to = "uid")]

46 pub creator: BelongsTo<super::super::users::user::Entity>,

47}

48

49impl ActiveModelBehavior for ActiveModel {}

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

@@ -0,0 +1 @@

1pub mod job;

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

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

77pub mod user_email_change_tickets;

88pub mod user_email_verification_codes;

99pub mod user_emails;

10pub mod user_external_identity;

1011pub mod user_favorite_memes;

1112pub mod user_followers;

1213pub mod user_notifications;

Added lib/models/src/users/user_external_identity.rs +29 −0 View lib/models/src/users/user_external_identity.rs

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

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 = "user_external_identity")]

8pub struct Model {

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

10 pub id: Uuid,

11 #[sea_orm(unique_key = "provider_login")]

12 pub provider: String,

13 pub external_id: Option<String>,

14 #[sea_orm(unique_key = "provider_login")]

15 pub lower_login: String,

16 pub login: String,

17 pub name: Option<String>,

18 #[sea_orm(indexed)]

19 pub email: Option<String>,

20 pub avatar_url: Option<String>,

21 #[sea_orm(indexed)]

22 pub user_id: Option<Uuid>,

23 pub created_at: OffsetDateTime,

24 pub updated_at: OffsetDateTime,

25 #[sea_orm(belongs_to, from = "user_id", to = "uid")]

26 pub user: BelongsTo<Option<super::user::Entity>>,

27}

28

29impl ActiveModelBehavior for ActiveModel {}

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

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

22pub mod common;

33pub mod dir;

44pub mod git;

5pub mod import;

56pub mod issue;

67pub mod migrate;

78pub mod products;

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

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

5858 let search =

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

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

61 let import = ::import::ImportService::new(state.clone());

6162 let health_state = state;

6263

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

@@ -68,10 +69,12 @@

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

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

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

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

72 .merge(crate::search::router().with_state(search.clone()))

73 .merge(::import::routes::router(import.clone()).with_state(import.clone()));

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

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

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

77 .merge(::import::routes::internal_router().with_state(import))

7578 .route_layer(axum::middleware::from_fn_with_state(

7679 auth.clone(),

7780 auth::share::guard::admin_guard,

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

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

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

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

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

115 m!("import", import::openapi::ImportApiDoc::openapi());

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

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

117118 doc.info.license = None;

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

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

1212space = { workspace = true }

1313user = { workspace = true }

1414git = { workspace = true }

15import = { workspace = true }

1516search = { workspace = true }

1617export = { workspace = true }

1718session = { workspace = true }

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

@@ -4534,6 +4534,32 @@

45344534]

45354535

45364536[[package]]

4537name = "import"

4538version = "1.0.0"

4539dependencies = [

4540 "async-trait",

4541 "auth",

4542 "axum",

4543 "cache",

4544 "config",

4545 "export",

4546 "git",

4547 "models",

4548 "reqwest 0.12.28",

4549 "sea-orm",

4550 "serde",

4551 "serde_json",

4552 "session",

4553 "thiserror 2.0.20",

4554 "time 0.3.55",

4555 "tokio",

4556 "tokio-util",

4557 "tracing",

4558 "utoipa",

4559 "uuid",

4560]

4561

4562[[package]]

45374563name = "indexmap"

45384564version = "2.14.0"

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

@@ -8227,6 +8253,7 @@

82278253 "config",

82288254 "export",

82298255 "git",

8256 "import",

82308257 "search",

82318258 "tokio",

82328259 "tokio-util",

@@ -9215,6 +9242,7 @@

92159242 "config",

92169243 "export",

92179244 "git",

9245 "import",

92189246 "issues",

92199247 "opentelemetry",

92209248 "opentelemetry-otlp",

Modified Cargo.toml +1 −0 View Cargo.toml

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

2727space = { path = "lib/space" }

2828issues = { path = "lib/issues" }

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

30import = { path = "lib/import" }

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

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

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

Modified openapi.json View openapi.json

Diff too large to display.