gitdataai/lib/service/git/fork.rs

use db::sqlx;
use git::rpc::{proto as p, proto::fork_service_client::ForkServiceClient};
use model::repos::RepoModel;
use serde::{Deserialize, Serialize};
use session::Session;

use crate::{
    AppService, Pagination, error::AppError, metrics::with_op_metric,
    session_user,
};

#[derive(Debug, Clone, Serialize, Deserialize, utoipa::ToSchema)]
pub struct ForkResponse {
    #[schema(value_type = String)]
    pub id: uuid::Uuid,
    pub name: String,
    pub description: Option<String>,
    pub default_branch: String,
    pub visibility: String,
    #[schema(value_type = String)]
    pub source_repo: uuid::Uuid,
    #[schema(value_type = String)]
    pub forked_by: uuid::Uuid,
    #[schema(value_type = String)]
    pub created_at: chrono::DateTime<chrono::Utc>,
}

#[derive(Debug, Clone, Deserialize, utoipa::ToSchema)]
pub struct CreateFork {
    pub name: Option<String>,
    pub visibility: Option<String>,
}

#[derive(db::sqlx::FromRow)]
struct ForkListRow {
    source_repo: uuid::Uuid,
    forked_by: uuid::Uuid,
    fork_created_at: chrono::DateTime<chrono::Utc>,
    repo_id: uuid::Uuid,
    repo_name: String,
    repo_description: Option<String>,
    repo_default_branch: String,
    repo_visibility: String,
}

impl AppService {
    #[tracing::instrument(skip(self, ctx), fields(workspace = %wk_name, repo = %repo_name))]
    pub async fn repo_fork_create(
        &self,
        ctx: &Session,
        wk_name: &str,
        repo_name: &str,
        params: CreateFork,
    ) -> Result<ForkResponse, AppError> {
        with_op_metric(&self.metrics.repo_fork_total, &[], async {
            let user_uid = session_user(ctx)?;
        let src_wk = self.workspace_resolve(wk_name).await?;
        self.workspace_require_member(src_wk.id, user_uid).await?;
        let source_repo = self.repo_resolve(src_wk.id, repo_name).await?;

        if source_repo.visibility == "private" {
            return Err(AppError::Forbidden(
                "cannot fork a private repo".to_string(),
            ));
        }

        let fork_name = params.name.unwrap_or_else(|| source_repo.name.clone());
        let fork_visibility = params
            .visibility
            .unwrap_or_else(|| source_repo.visibility.clone());

        let existing = sqlx::query_scalar::<_, bool>(
            "SELECT EXISTS(SELECT 1 FROM repo WHERE wk = $1 AND name = $2 AND deleted_at IS NULL AND created_by = $3)",
        )
        .bind(src_wk.id)
        .bind(&fork_name)
        .bind(user_uid)
        .fetch_one(self.db.reader())
        .await
        .map_err(|e| AppError::DatabaseError(e.to_string()))?;
        if existing {
            return Err(AppError::Conflict("fork already exists".to_string()));
        }

        let repo_id = uuid::Uuid::now_v7();
        let now = chrono::Utc::now();
        let description = source_repo.description.clone();
        let default_branch = source_repo.default_branch.clone();

        let mut txn = self.db.begin().await.map_err(|_| AppError::TxnError)?;
        let _fork_repo = sqlx::query_as::<_, RepoModel>(
            "INSERT INTO repo (id, wk, name, description, default_branch, visibility, size_bytes, \
             is_archived, is_template, is_mirror, created_by, storage_path, created_at, updated_at) \
             VALUES ($1, $2, $3, $4, $5, $6, 0, false, false, false, $7, '', $8, $8) \
             RETURNING id, wk, name, description, default_branch, visibility, size_bytes, \
             is_archived, is_template, is_mirror, created_by, storage_path, created_at, updated_at, deleted_at",
        )
        .bind(repo_id)
        .bind(src_wk.id)
        .bind(&fork_name)
        .bind(&description)
        .bind(&default_branch)
        .bind(&fork_visibility)
        .bind(user_uid)
        .bind(now)
        .fetch_one(&mut **txn.inner_mut())
        .await
        .map_err(|e| AppError::DatabaseError(e.to_string()))?;
        let storage_root = self
            .config
            .repos_root()
            .map_err(|e| AppError::InternalServerError(e.to_string()))?;

        let mut client = ForkServiceClient::new(self.git.clone());
        let rpc_resp = client
            .fork_bare(tonic::Request::new(p::ForkBareRequest {
                storage_root,
                source_storage_path: source_repo.storage_path.clone(),
                params: Some(p::ForkRepoParams {
                    namespace: src_wk.name.clone(),
                    repo_name: fork_name.clone(),
                    default_branch: default_branch.clone(),
                    description: description.clone(),
                    enable_lfs: false,
                }),
            }))
            .await
            .map_err(crate::git::rpc_err)?
            .into_inner();
        let fork_repo = sqlx::query_as::<_, RepoModel>(
            "UPDATE repo SET storage_path = $1 WHERE id = $2 \
             RETURNING id, wk, name, description, default_branch, visibility, size_bytes, \
             is_archived, is_template, is_mirror, created_by, storage_path, created_at, updated_at, deleted_at",
        )
        .bind(&rpc_resp.storage_path)
        .bind(repo_id)
        .fetch_one(&mut **txn.inner_mut())
        .await
        .map_err(|e| AppError::DatabaseError(e.to_string()))?;
        sqlx::query(
            "INSERT INTO repo_fork (id, repo, source_repo, forked_by, created_at) \
             VALUES ($1, $2, $3, $4, $5)",
        )
        .bind(uuid::Uuid::now_v7())
        .bind(fork_repo.id)
        .bind(source_repo.id)
        .bind(user_uid)
        .bind(now)
        .execute(&mut **txn.inner_mut())
        .await
        .map_err(|e| AppError::DatabaseError(e.to_string()))?;

        txn.commit().await.map_err(|_| AppError::TxnError)?;
        self.queue_sync(repo_id).await;

            Ok(ForkResponse {
            id: fork_repo.id,
            name: fork_repo.name,
            description: fork_repo.description,
            default_branch: fork_repo.default_branch,
            visibility: fork_repo.visibility,
            source_repo: source_repo.id,
            forked_by: user_uid,
            created_at: fork_repo.created_at,
            })
        }).await
    }

    pub async fn repo_fork_list(
        &self,
        ctx: &Session,
        wk_name: &str,
        repo_name: &str,
        pagination: Pagination,
    ) -> Result<Vec<ForkResponse>, AppError> {
        let repo = self.git_require_member(ctx, wk_name, repo_name).await?;

        let rows = sqlx::query_as::<_, ForkListRow>(
            "SELECT f.id as fork_id, f.source_repo, f.forked_by, f.created_at as fork_created_at, \
             r.id as repo_id, r.name as repo_name, r.description as repo_description, \
             r.default_branch as repo_default_branch, r.visibility as repo_visibility, \
             r.created_at as repo_created_at \
             FROM repo_fork f \
             INNER JOIN repo r ON r.id = f.repo AND r.deleted_at IS NULL \
             WHERE f.source_repo = $1 \
             ORDER BY f.created_at DESC \
             OFFSET $2 LIMIT $3",
        )
        .bind(repo.id)
        .bind(pagination.offset() as i64)
        .bind(pagination.limit() as i64)
        .fetch_all(self.db.reader())
        .await
        .map_err(|e| AppError::DatabaseError(e.to_string()))?;

        Ok(rows
            .into_iter()
            .map(|row| ForkResponse {
                id: row.repo_id,
                name: row.repo_name,
                description: row.repo_description,
                default_branch: row.repo_default_branch,
                visibility: row.repo_visibility,
                source_repo: row.source_repo,
                forked_by: row.forked_by,
                created_at: row.fork_created_at,
            })
            .collect())
    }
}