563bb99
use config::AppConfig;
use config::import::ImportConfig;
use export::AppState;
use export::pagination::Page;
use git::service::GitService;
use git::service::repos::CreateRepoCmd;
use models::import::job;
use models::space::{space_members, spaces};
use sea_orm::{
ActiveModelTrait, ColumnTrait, DatabaseConnection, EntityTrait, PaginatorTrait, QueryFilter,
QueryOrder, QuerySelect, Set,
};
use serde::{Deserialize, Serialize};
use time::OffsetDateTime;
use utoipa::{IntoParams, ToSchema};
use uuid::Uuid;
use crate::errors::ImportError;
use crate::providers::{self, ImportProvider, ProbeInfo};
#[derive(Debug, Clone, Serialize, Deserialize, Default, PartialEq)]
pub struct ImportOptions {
pub owner: String,
pub owner_external_id: Option<String>,
pub owner_avatar_url: Option<String>,
pub source_default_branch: Option<String>,
pub source_size_bytes: Option<i64>,
pub source_is_private: bool,
pub credential_username: Option<String>,
pub enable_lfs: bool,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, ToSchema)]
#[serde(rename_all = "snake_case")]
pub enum ImportJobState {
Queued,
Processing,
Succeeded,
Failed,
Canceled,
}
impl ImportJobState {
#[must_use]
pub fn as_str(self) -> &'static str {
match self {
Self::Queued => "queued",
Self::Processing => "processing",
Self::Succeeded => "succeeded",
Self::Failed => "failed",
Self::Canceled => "canceled",
}
}
#[must_use]
pub fn parse(value: &str) -> Self {
match value {
"processing" => Self::Processing,
"succeeded" => Self::Succeeded,
"failed" => Self::Failed,
"canceled" => Self::Canceled,
_ => Self::Queued,
}
}
#[must_use]
pub fn is_terminal(self) -> bool {
matches!(self, Self::Succeeded | Self::Failed | Self::Canceled)
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, ToSchema)]
#[serde(rename_all = "snake_case")]
pub enum ImportStage {
Probe,
Clone,
Sync,
Finalize,
Done,
}
impl ImportStage {
#[must_use]
pub fn as_str(self) -> &'static str {
match self {
Self::Probe => "probe",
Self::Clone => "clone",
Self::Sync => "sync",
Self::Finalize => "finalize",
Self::Done => "done",
}
}
#[must_use]
pub fn parse(value: &str) -> Self {
match value {
"clone" => Self::Clone,
"sync" => Self::Sync,
"finalize" => Self::Finalize,
"done" => Self::Done,
_ => Self::Probe,
}
}
}
#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct ImportProbeDto {
pub provider: ImportProvider,
pub owner: String,
pub name: String,
pub clone_url: String,
pub default_branch: String,
pub is_private: bool,
pub description: Option<String>,
pub size_bytes: Option<i64>,
}
impl From<&ProbeInfo> for ImportProbeDto {
fn from(info: &ProbeInfo) -> Self {
Self {
provider: info.provider,
owner: info.owner.clone(),
name: info.name.clone(),
clone_url: info.clone_url.clone(),
default_branch: info.default_branch.clone(),
is_private: info.is_private,
description: info.description.clone(),
size_bytes: info.size_bytes,
}
}
}
#[derive(Debug, Clone, Default, Serialize, Deserialize, ToSchema, PartialEq)]
pub struct ImportStatsDto {
pub branches: i64,
pub tags: i64,
pub commits: i64,
pub size_bytes: i64,
pub default_branch: Option<String>,
pub warnings: Vec<String>,
}
#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct ImportJobDto {
pub id: Uuid,
pub provider: ImportProvider,
pub source_url: String,
pub base_url: Option<String>,
pub clone_url: String,
pub space_id: Uuid,
pub space_name: String,
pub repo_id: Option<Uuid>,
pub repo_name: String,
pub state: ImportJobState,
pub stage: ImportStage,
pub attempts: i32,
pub last_error: Option<String>,
pub stats: ImportStatsDto,
pub created_by: Uuid,
pub created_at: OffsetDateTime,
pub updated_at: OffsetDateTime,
pub started_at: Option<OffsetDateTime>,
pub finished_at: Option<OffsetDateTime>,
}
pub type ImportJobPage = Page<ImportJobDto>;
#[derive(Debug, Clone, Deserialize, ToSchema)]
pub struct ProbeImportCmd {
pub provider: ImportProvider,
pub source: String,
#[serde(default)]
pub base_url: Option<String>,
#[serde(default)]
pub token: Option<String>,
}
#[derive(Debug, Clone, Deserialize, ToSchema)]
pub struct CreateImportCmd {
pub provider: ImportProvider,
pub source: String,
#[serde(default)]
pub base_url: Option<String>,
#[serde(default)]
pub token: Option<String>,
#[serde(default)]
pub name: Option<String>,
#[serde(default)]
pub description: Option<String>,
#[serde(default)]
pub is_private: Option<bool>,
#[serde(default)]
pub parent_dir_id: Option<Uuid>,
#[serde(default)]
pub enable_lfs: bool,
}
#[derive(Debug, Clone, Deserialize, ToSchema)]
pub struct RetryImportCmd {
#[serde(default)]
pub token: Option<String>,
}
#[derive(Debug, Clone, Deserialize, IntoParams, ToSchema)]
pub struct ListImportQuery {
pub state: Option<String>,
pub repo_name: Option<String>,
pub limit: Option<u64>,
pub offset: Option<u64>,
}
#[derive(Debug, Clone, Serialize)]
pub struct RequeueResultDto {
pub failed: u64,
pub stale: u64,
}
#[derive(Debug, Clone, Deserialize, IntoParams, ToSchema)]
pub struct ImportSpacePath {
pub space_name: String,
}
#[derive(Debug, Clone, Deserialize, IntoParams, ToSchema)]
pub struct ImportJobPath {
pub space_name: String,
pub job_id: Uuid,
}
#[derive(Clone)]
pub struct ImportService {
pub state: AppState,
pub git: GitService,
}
impl ImportService {
pub fn new(state: AppState) -> Self {
let git = GitService::new(state.clone());
Self { state, git }
}
pub fn config(&self) -> &AppConfig {
&self.state.config
}
pub fn import_config(&self) -> &ImportConfig {
&self.state.config.import
}
pub fn db_read(&self) -> &DatabaseConnection {
self.state.database.read()
}
pub fn db_write(&self) -> &DatabaseConnection {
self.state.database.write()
}
fn is_platform_admin(&self, caller: Uuid) -> bool {
self.config().apihost.is_platform_admin(&caller.to_string())
}
pub fn dto(&self, row: &job::Model, space_name: &str) -> ImportJobDto {
ImportJobDto {
id: row.id,
provider: ImportProvider::parse(&row.provider).unwrap_or(ImportProvider::Github),
source_url: row.source_url.clone(),
base_url: row.base_url.clone(),
clone_url: row.clone_url.clone(),
space_id: row.space_id,
space_name: space_name.to_string(),
repo_id: row.repo_id,
repo_name: row.repo_name.clone(),
state: ImportJobState::parse(&row.state),
stage: ImportStage::parse(&row.stage),
attempts: row.attempts,
last_error: row.last_error.clone(),
stats: serde_json::from_value(row.stats.clone()).unwrap_or_default(),
created_by: row.created_by,
created_at: row.created_at,
updated_at: row.updated_at,
started_at: row.started_at,
finished_at: row.finished_at,
}
}
pub async fn lookup_space(&self, space_name: &str) -> Result<spaces::Model, ImportError> {
let lower = space_name.trim().to_lowercase();
if lower.is_empty() {
return Err(ImportError::NotFound);
}
spaces::Entity::find()
.filter(spaces::Column::LowerName.eq(&lower))
.filter(spaces::Column::IsActive.eq(true))
.one(self.db_read())
.await?
.ok_or(ImportError::NotFound)
}
async fn require_member(&self, space: &spaces::Model, caller: Uuid) -> Result<(), ImportError> {
if space.owner_id == caller || self.is_platform_admin(caller) {
return Ok(());
}
let member = space_members::Entity::find_by_id((space.id, caller))
.one(self.db_read())
.await?;
let active = member
.is_some_and(|row| row.status == models::space::space_members::MemberStatus::Active);
if active {
Ok(())
} else {
Err(ImportError::Forbidden)
}
}
pub async fn probe_source(
&self,
provider: ImportProvider,
source: &str,
base_url: Option<&str>,
token: Option<&str>,
) -> Result<ProbeInfo, ImportError> {
let config = self.import_config();
if !config.enabled {
return Err(ImportError::Disabled);
}
if let Some(value) = base_url.map(str::trim).filter(|value| !value.is_empty()) {
providers::ensure_url_allowed(value, config)?;
}
providers::probe(providers::ProbeRequest {
provider,
source,
base_url,
token,
config,
})
.await
}
pub async fn probe(
&self,
caller: Uuid,
space_name: &str,
cmd: ProbeImportCmd,
) -> Result<ImportProbeDto, ImportError> {
let space = self.lookup_space(space_name).await?;
self.require_member(&space, caller).await?;
let info = self
.probe_source(
cmd.provider,
&cmd.source,
cmd.base_url.as_deref(),
cmd.token.as_deref(),
)
.await?;
providers::ensure_url_allowed(&info.clone_url, self.import_config())?;
Ok(ImportProbeDto::from(&info))
}
pub async fn create(
&self,
caller: Uuid,
space_name: &str,
cmd: CreateImportCmd,
) -> Result<ImportJobDto, ImportError> {
let config = self.import_config().clone();
if !config.enabled {
return Err(ImportError::Disabled);
}
let space = self.lookup_space(space_name).await?;
if space.owner_id != caller && !self.is_platform_admin(caller) {
return Err(ImportError::Forbidden);
}
let info = self
.probe_source(
cmd.provider,
&cmd.source,
cmd.base_url.as_deref(),
cmd.token.as_deref(),
)
.await?;
providers::ensure_url_allowed(&info.clone_url, &config)?;
if config.max_repo_bytes > 0
&& info.size_bytes.unwrap_or(0).max(0) as u64 > config.max_repo_bytes
{
return Err(ImportError::BadRequest(format!(
"source repository exceeds the {} byte import limit",
config.max_repo_bytes
)));
}
let repo_name = match cmd.name.as_deref().map(str::trim).filter(|v| !v.is_empty()) {
Some(name) => name.to_string(),
None => info.name.clone(),
};
let description = cmd
.description
.clone()
.filter(|value| !value.trim().is_empty())
.or_else(|| info.description.clone());
let is_private = cmd.is_private.unwrap_or(info.is_private);
let init_message = format!("Import {} from {}", repo_name, info.clone_url);
let repo = self
.git
.create_repo(
space.name.clone(),
caller,
CreateRepoCmd {
name: repo_name.clone(),
description,
is_private,
default_branch: Some(info.default_branch.clone()),
parent_dir_id: cmd.parent_dir_id,
enable_lfs: cmd.enable_lfs,
license_name: None,
init_message: Some(init_message),
},
)
.await?;
let token_enc = match cmd
.token
.as_deref()
.map(str::trim)
.filter(|v| !v.is_empty())
{
Some(token) => Some(encrypt_token(&config.token_secret, token)?),
None => None,
};
let options = ImportOptions {
owner: info.owner.clone(),
owner_external_id: info.owner_external_id.clone(),
owner_avatar_url: info.owner_avatar_url.clone(),
source_default_branch: Some(info.default_branch.clone()),
source_size_bytes: info.size_bytes,
source_is_private: info.is_private,
credential_username: info.credential_username.clone(),
enable_lfs: cmd.enable_lfs,
};
let now = OffsetDateTime::now_utc();
let row = job::ActiveModel {
id: Set(Uuid::now_v7()),
provider: Set(info.provider.as_str().to_string()),
source_url: Set(cmd.source.trim().to_string()),
base_url: Set(cmd
.base_url
.as_deref()
.map(str::trim)
.filter(|v| !v.is_empty())
.map(str::to_string)),
clone_url: Set(info.clone_url.clone()),
space_id: Set(space.id),
repo_id: Set(Some(repo.id)),
repo_name: Set(repo_name),
created_by: Set(caller),
state: Set(ImportJobState::Queued.as_str().to_string()),
stage: Set(ImportStage::Clone.as_str().to_string()),
attempts: Set(0),
last_error: Set(None),
token_enc: Set(token_enc),
options: Set(serde_json::to_value(&options).unwrap_or_default()),
stats: Set(serde_json::to_value(ImportStatsDto::default()).unwrap_or_default()),
run_after: Set(now),
locked_at: Set(None),
started_at: Set(None),
finished_at: Set(None),
created_at: Set(now),
updated_at: Set(now),
}
.insert(self.db_write())
.await?;
Ok(self.dto(&row, &space.name))
}
pub async fn list(
&self,
caller: Uuid,
space_name: &str,
query: ListImportQuery,
) -> Result<ImportJobPage, ImportError> {
let space = self.lookup_space(space_name).await?;
self.require_member(&space, caller).await?;
let limit = query.limit.unwrap_or(20).clamp(1, 100);
let offset = query.offset.unwrap_or(0);
let mut select = job::Entity::find().filter(job::Column::SpaceId.eq(space.id));
if space.owner_id != caller && !self.is_platform_admin(caller) {
select = select.filter(job::Column::CreatedBy.eq(caller));
}
if let Some(state) = query
.state
.as_deref()
.map(str::trim)
.filter(|v| !v.is_empty())
{
select = select.filter(job::Column::State.eq(state.to_string()));
}
if let Some(repo) = query
.repo_name
.as_deref()
.map(str::trim)
.filter(|v| !v.is_empty())
{
select = select.filter(job::Column::RepoName.eq(repo.to_string()));
}
let total = select.clone().count(self.db_read()).await? as i64;
let rows = select
.order_by_desc(job::Column::CreatedAt)
.limit(limit)
.offset(offset)
.all(self.db_read())
.await?;
let items = rows.iter().map(|row| self.dto(row, &space.name)).collect();
Ok(Page {
items,
total,
limit,
offset,
})
}
pub async fn load_job(
&self,
caller: Uuid,
space_name: &str,
job_id: Uuid,
) -> Result<(spaces::Model, job::Model), ImportError> {
let space = self.lookup_space(space_name).await?;
self.require_member(&space, caller).await?;
let row = job::Entity::find_by_id(job_id)
.filter(job::Column::SpaceId.eq(space.id))
.one(self.db_read())
.await?
.ok_or(ImportError::NotFound)?;
if row.created_by != caller && space.owner_id != caller && !self.is_platform_admin(caller) {
return Err(ImportError::Forbidden);
}
Ok((space, row))
}
pub async fn get(
&self,
caller: Uuid,
space_name: &str,
job_id: Uuid,
) -> Result<ImportJobDto, ImportError> {
let (space, row) = self.load_job(caller, space_name, job_id).await?;
Ok(self.dto(&row, &space.name))
}
pub async fn requeue_pending(&self) -> Result<RequeueResultDto, ImportError> {
let now = OffsetDateTime::now_utc();
let stale_cutoff = now
- time::Duration::seconds(
i64::try_from(self.import_config().stale_lock_secs.max(1)).unwrap_or(1800),
);
let failed = job::Entity::update_many()
.col_expr(
job::Column::State,
sea_orm::sea_query::Expr::value(ImportJobState::Queued.as_str()),
)
.col_expr(
job::Column::Stage,
sea_orm::sea_query::Expr::value(ImportStage::Clone.as_str()),
)
.col_expr(job::Column::Attempts, sea_orm::sea_query::Expr::value(0i32))
.col_expr(
job::Column::LockedAt,
sea_orm::sea_query::Expr::value(Option::<OffsetDateTime>::None),
)
.col_expr(
job::Column::FinishedAt,
sea_orm::sea_query::Expr::value(Option::<OffsetDateTime>::None),
)
.col_expr(job::Column::RunAfter, sea_orm::sea_query::Expr::value(now))
.col_expr(job::Column::UpdatedAt, sea_orm::sea_query::Expr::value(now))
.filter(job::Column::State.eq(ImportJobState::Failed.as_str()))
.exec(self.db_write())
.await?;
let stale = job::Entity::update_many()
.col_expr(
job::Column::State,
sea_orm::sea_query::Expr::value(ImportJobState::Queued.as_str()),
)
.col_expr(
job::Column::LockedAt,
sea_orm::sea_query::Expr::value(Option::<OffsetDateTime>::None),
)
.col_expr(job::Column::RunAfter, sea_orm::sea_query::Expr::value(now))
.col_expr(job::Column::UpdatedAt, sea_orm::sea_query::Expr::value(now))
.filter(job::Column::State.eq(ImportJobState::Processing.as_str()))
.filter(job::Column::LockedAt.lt(stale_cutoff))
.exec(self.db_write())
.await?;
Ok(RequeueResultDto {
failed: failed.rows_affected,
stale: stale.rows_affected,
})
}
pub async fn cancel(
&self,
caller: Uuid,
space_name: &str,
job_id: Uuid,
) -> Result<ImportJobDto, ImportError> {
let (space, row) = self.load_job(caller, space_name, job_id).await?;
let state = ImportJobState::parse(&row.state);
if state.is_terminal() {
return Err(ImportError::Conflict(format!(
"import job is already {}",
state.as_str()
)));
}
let now = OffsetDateTime::now_utc();
let updated = job::ActiveModel {
id: Set(row.id),
state: Set(ImportJobState::Canceled.as_str().to_string()),
token_enc: Set(None),
locked_at: Set(None),
finished_at: Set(Some(now)),
updated_at: Set(now),
..Default::default()
}
.update(self.db_write())
.await?;
Ok(self.dto(&updated, &space.name))
}
pub async fn retry(
&self,
caller: Uuid,
space_name: &str,
job_id: Uuid,
cmd: RetryImportCmd,
) -> Result<ImportJobDto, ImportError> {
let (space, row) = self.load_job(caller, space_name, job_id).await?;
let state = ImportJobState::parse(&row.state);
if !matches!(state, ImportJobState::Failed | ImportJobState::Canceled) {
return Err(ImportError::Conflict(format!(
"only failed or canceled import jobs can be retried, current state is {}",
state.as_str()
)));
}
let config = self.import_config().clone();
let token_enc = match cmd
.token
.as_deref()
.map(str::trim)
.filter(|v| !v.is_empty())
{
Some(token) => Some(encrypt_token(&config.token_secret, token)?),
None => None,
};
let now = OffsetDateTime::now_utc();
let updated = job::ActiveModel {
id: Set(row.id),
state: Set(ImportJobState::Queued.as_str().to_string()),
stage: Set(ImportStage::Clone.as_str().to_string()),
attempts: Set(0),
last_error: Set(None),
token_enc: Set(token_enc),
run_after: Set(now),
locked_at: Set(None),
finished_at: Set(None),
updated_at: Set(now),
..Default::default()
}
.update(self.db_write())
.await?;
Ok(self.dto(&updated, &space.name))
}
}
pub fn encrypt_token(secret: &str, token: &str) -> Result<String, ImportError> {
auth::share::crypto::aes_encrypt(secret, token)
.map_err(|e| ImportError::Internal(format!("token encryption failed: {e}")))
}
pub fn decrypt_token(secret: &str, stored: &str) -> Result<String, ImportError> {
auth::share::crypto::aes_decrypt(secret, stored)
.map_err(|e| ImportError::Internal(format!("token decryption failed: {e}")))
}
#[async_trait::async_trait]
impl session::SessionGuard for ImportService {
async fn lookup_session_with_user(
&self,
token_hash: &str,
) -> Result<Option<session::SessionWithUser>, session::SessionError> {
session::lookup_session_with_user(self.db_read(), token_hash).await
}
}