use serde::{Deserialize, Serialize}; use uuid::Uuid; use crate::error::AppError; use crate::models::common::{GitService, Role, Visibility}; use crate::models::repos::Repo; use crate::service::RepoService; use crate::session::Session; use super::util::{ clamp_limit_offset, ensure_affected, merge_optional_text, parse_enum, required_text, }; #[derive(Deserialize, Serialize, Clone, Debug, utoipa::ToSchema)] pub struct CreateRepoParams { pub name: String, pub description: Option, pub visibility: Option, pub default_branch: Option, pub git_service: Option, pub storage_node_ids: Option>, pub storage_path: Option, } #[derive(Deserialize, Serialize, Clone, Debug, utoipa::ToSchema)] pub struct UpdateRepoParams { pub name: Option, pub description: Option, pub visibility: Option, pub default_branch: Option, } fn validate_storage_path(path: &str) -> Result { let path = path.trim().trim_matches('/'); if path.is_empty() || path.contains("..") || path.split('/').any(|part| part.is_empty() || part == ".") { return Err(AppError::BadRequest("storage_path is invalid".into())); } Ok(path.to_string()) } impl RepoService { pub async fn repo_list( &self, ctx: &Session, wk_name: &str, limit: i64, offset: i64, ) -> Result, AppError> { let user_uid = ctx.user().ok_or(AppError::Unauthorized)?; let ws = self.resolve_workspace(wk_name).await?; self.ensure_workspace_readable(user_uid, &ws).await?; let workspace_id = ws.id; let (limit, offset) = clamp_limit_offset(limit, offset); sqlx::query_as::<_, Repo>( "SELECT id, workspace_id, owner_id, name, description, default_branch, visibility, status, is_fork, forked_from_repo_id, storage_node_ids, primary_storage_node_id, storage_path, git_service, archived_at, created_at, updated_at, deleted_at \ FROM repo WHERE workspace_id = $1 AND deleted_at IS NULL AND (owner_id = $2 OR visibility = 'public' OR id IN (SELECT repo_id FROM repo_member WHERE user_id = $2 AND status = 'active') OR (visibility = 'internal' AND EXISTS (SELECT 1 FROM workspace_member WHERE workspace_id = $1 AND user_id = $2 AND status = 'active'))) \ ORDER BY created_at DESC LIMIT $3 OFFSET $4", ) .bind(workspace_id) .bind(user_uid) .bind(limit) .bind(offset) .fetch_all(self.ctx.db.reader()) .await .map_err(AppError::Database) } pub async fn repo_get( &self, ctx: &Session, wk_name: &str, repo_name: &str, ) -> Result { let user_uid = ctx.user().ok_or(AppError::Unauthorized)?; let repo = self.resolve_repo(wk_name, repo_name).await?; self.ensure_repo_readable(user_uid, &repo).await?; Ok(repo) } pub async fn repo_create( &self, ctx: &Session, wk_name: &str, params: CreateRepoParams, ) -> Result { let user_uid = ctx.user().ok_or(AppError::Unauthorized)?; let ws = self.resolve_workspace(wk_name).await?; let workspace_id = ws.id; self.ensure_workspace_role_at_least(user_uid, &ws, Role::Member) .await?; let name = required_text(params.name, "name")?; let visibility = match params.visibility { Some(ref v) => parse_enum( Some(v.clone()), Visibility::Private, Visibility::Unknown, "visibility", )?, None => { let settings_visibility: String = sqlx::query_scalar( "SELECT default_repo_visibility FROM workspace_settings WHERE workspace_id = $1", ) .bind(workspace_id) .fetch_one(self.ctx.db.reader()) .await .map_err(AppError::Database)?; settings_visibility.parse().unwrap_or(Visibility::Private) } }; if visibility == Visibility::Public { let allow_public_repos: bool = sqlx::query_scalar( "SELECT allow_public_repos FROM workspace_settings WHERE workspace_id = $1", ) .bind(workspace_id) .fetch_one(self.ctx.db.reader()) .await .map_err(AppError::Database)?; if !allow_public_repos { return Err(AppError::BadRequest( "public repositories are disabled for this workspace".into(), )); } } let default_branch = required_text( params.default_branch.unwrap_or_else(|| "main".to_string()), "default_branch", )?; let git_service = match params.git_service { Some(ref v) => parse_enum( Some(v.clone()), GitService::Local, GitService::Unknown, "git_service", )?, None => GitService::Local, }; let available_storage_nodes: std::collections::HashSet = self.ctx.registry.git_node_ids().into_iter().collect(); if available_storage_nodes.is_empty() { return Err(AppError::Config("no git storage nodes configured".into())); } let storage_node_ids = params.storage_node_ids.unwrap_or_else(|| { available_storage_nodes .iter() .copied() .collect::>() }); if storage_node_ids.is_empty() || storage_node_ids .iter() .any(|node_id| !available_storage_nodes.contains(node_id)) { return Err(AppError::BadRequest("invalid storage_node_ids".into())); } let primary_storage_node_id = storage_node_ids[0]; let now = chrono::Utc::now(); let repo_id = Uuid::now_v7(); let storage_path = match params.storage_path { Some(path) if !path.trim().is_empty() => validate_storage_path(&path)?, Some(_) => return Err(AppError::BadRequest("storage_path is invalid".into())), None => format!("repos/{}/{}", workspace_id, repo_id), }; let mut txn = self .ctx .db .writer() .begin() .await .map_err(|_| AppError::TxnError)?; sqlx::query("SET LOCAL app.current_user_id = $1") .bind(user_uid) .execute(&mut *txn) .await .map_err(AppError::Database)?; let repo = sqlx::query_as::<_, Repo>( "INSERT INTO repo (id, workspace_id, owner_id, name, description, default_branch, \ visibility, status, is_fork, forked_from_repo_id, storage_node_ids, \ primary_storage_node_id, storage_path, git_service, archived_at, created_at, updated_at) \ VALUES ($1, $2, $3, $4, $5, $6, $7, 'active', false, NULL, $8, $9, $10, $11, NULL, $12, $12) \ RETURNING id, workspace_id, owner_id, name, description, default_branch, visibility, status, is_fork, forked_from_repo_id, storage_node_ids, primary_storage_node_id, storage_path, git_service, archived_at, created_at, updated_at, deleted_at", ) .bind(repo_id) .bind(workspace_id) .bind(user_uid) .bind(&name) .bind(params.description.as_deref()) .bind(&default_branch) .bind(visibility) .bind(&storage_node_ids) .bind(primary_storage_node_id) .bind(&storage_path) .bind(git_service) .bind(now) .fetch_one(&mut *txn) .await .map_err(AppError::Database)?; sqlx::query( "INSERT INTO repo_member (id, repo_id, user_id, role, status, joined_at, created_at, updated_at) \ VALUES ($1, $2, $3, 'owner', 'active', $4, $4, $4)", ) .bind(Uuid::now_v7()) .bind(repo_id) .bind(user_uid) .bind(now) .execute(&mut *txn) .await .map_err(AppError::Database)?; sqlx::query( "INSERT INTO repo_stats (repo_id, stars_count, watchers_count, forks_count, branches_count, \ tags_count, commits_count, releases_count, open_issues_count, open_pull_requests_count, \ size_bytes, updated_at) \ VALUES ($1, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0, $2)", ) .bind(repo_id) .bind(now) .execute(&mut *txn) .await .map_err(AppError::Database)?; sqlx::query( "INSERT INTO repo_branch (id, repo_id, name, commit_sha, protected, default_branch, created_by, created_at, updated_at) \ VALUES ($1, $2, $3, '', false, true, $4, $5, $5)", ) .bind(Uuid::now_v7()) .bind(repo_id) .bind(&default_branch) .bind(user_uid) .bind(now) .execute(&mut *txn) .await .map_err(AppError::Database)?; sqlx::query( "UPDATE workspace_stats SET repos_count = repos_count + 1, updated_at = $1 WHERE workspace_id = $2", ) .bind(now) .bind(workspace_id) .execute(&mut *txn) .await .map_err(AppError::Database)?; txn.commit().await.map_err(|_| AppError::TxnError)?; // Init git repo on primary node (nodes auto-sync). if let Some(mut client) = self.ctx.registry.get_git_client(&primary_storage_node_id) { let req = tonic::Request::new(crate::pb::repo::InitRepositoryRequest { repository: Some(crate::pb::repo::RepositoryHeader { storage_name: ws.name.clone(), relative_path: format!("{}.git", name), storage_path: storage_path.clone(), }), bare: true, object_format: crate::pb::repo::ObjectFormat::Sha1 as i32, initial_branch: default_branch.clone(), }); if let Err(err) = client.repository.init_repository(req).await { tracing::error!(repo_id = %repo_id, error = %err, "Failed to init git repo"); let _ = sqlx::query( "UPDATE repo SET status = 'deleted', deleted_at = $1 WHERE id = $2", ) .bind(chrono::Utc::now()) .bind(repo_id) .execute(self.ctx.db.writer()) .await; let _ = sqlx::query("UPDATE workspace_stats SET repos_count = GREATEST(repos_count - 1, 0), updated_at = $1 WHERE workspace_id = $2") .bind(chrono::Utc::now()).bind(workspace_id).execute(self.ctx.db.writer()).await; return Err(AppError::InternalServerError(err.to_string())); } } Ok(repo) } pub async fn repo_update( &self, ctx: &Session, wk_name: &str, repo_name: &str, params: UpdateRepoParams, ) -> Result { let user_uid = ctx.user().ok_or(AppError::Unauthorized)?; let repo = self.resolve_repo(wk_name, repo_name).await?; let repo_id = repo.id; self.ensure_repo_role_at_least(user_uid, &repo, Role::Admin) .await?; let name = merge_optional_text(params.name, Some(repo.name.clone())).unwrap_or(repo.name.clone()); let description = merge_optional_text(params.description, repo.description); let visibility = parse_enum( params.visibility, repo.visibility, Visibility::Unknown, "visibility", )?; if visibility == Visibility::Public { let allow_public_repos: bool = sqlx::query_scalar( "SELECT allow_public_repos FROM workspace_settings WHERE workspace_id = $1", ) .bind(repo.workspace_id) .fetch_one(self.ctx.db.reader()) .await .map_err(AppError::Database)?; if !allow_public_repos { return Err(AppError::BadRequest( "public repositories are disabled for this workspace".into(), )); } } let now = chrono::Utc::now(); let mut txn = self .ctx .db .writer() .begin() .await .map_err(|_| AppError::TxnError)?; sqlx::query("SET LOCAL app.current_user_id = $1") .bind(user_uid) .execute(&mut *txn) .await .map_err(AppError::Database)?; sqlx::query( "UPDATE repo SET name = $1, description = $2, visibility = $3, updated_at = $4 \ WHERE id = $5 AND deleted_at IS NULL", ) .bind(&name) .bind(&description) .bind(visibility) .bind(now) .bind(repo_id) .execute(&mut *txn) .await .map_err(AppError::Database)?; if let Some(ref new_default) = params.default_branch && new_default != &repo.default_branch { // Check if the branch exists let branch_exists: bool = sqlx::query_scalar( "SELECT EXISTS(SELECT 1 FROM repo_branch WHERE repo_id = $1 AND name = $2)", ) .bind(repo_id) .bind(new_default) .fetch_one(&mut *txn) .await .map_err(AppError::Database)?; if !branch_exists { return Err(AppError::BadRequest(format!( "Branch '{}' does not exist", new_default ))); } sqlx::query( "UPDATE repo_branch SET default_branch = false, updated_at = $1 WHERE repo_id = $2 AND default_branch = true", ) .bind(now) .bind(repo_id) .execute(&mut *txn) .await .map_err(AppError::Database)?; sqlx::query( "UPDATE repo_branch SET default_branch = true, updated_at = $1 WHERE repo_id = $2 AND name = $3", ) .bind(now) .bind(repo_id) .bind(new_default) .execute(&mut *txn) .await .map_err(AppError::Database)?; sqlx::query("UPDATE repo SET default_branch = $1, updated_at = $2 WHERE id = $3") .bind(new_default) .bind(now) .bind(repo_id) .execute(&mut *txn) .await .map_err(AppError::Database)?; } let result = sqlx::query_as::<_, Repo>( "SELECT id, workspace_id, owner_id, name, description, default_branch, visibility, status, is_fork, forked_from_repo_id, storage_node_ids, primary_storage_node_id, storage_path, git_service, archived_at, created_at, updated_at, deleted_at \ FROM repo WHERE id = $1 AND deleted_at IS NULL", ) .bind(repo_id) .fetch_one(&mut *txn) .await .map_err(AppError::Database)?; txn.commit().await.map_err(|_| AppError::TxnError)?; Ok(result) } pub async fn repo_archive( &self, ctx: &Session, wk_name: &str, repo_name: &str, ) -> Result<(), AppError> { let user_uid = ctx.user().ok_or(AppError::Unauthorized)?; let repo = self.resolve_repo(wk_name, repo_name).await?; let repo_id = repo.id; self.ensure_repo_role_at_least(user_uid, &repo, Role::Owner) .await?; let now = chrono::Utc::now(); let mut txn = self .ctx .db .writer() .begin() .await .map_err(|_| AppError::TxnError)?; sqlx::query("SET LOCAL app.current_user_id = $1") .bind(user_uid) .execute(&mut *txn) .await .map_err(AppError::Database)?; let result = sqlx::query( "UPDATE repo SET status = 'archived', archived_at = $1, updated_at = $1 WHERE id = $2 AND deleted_at IS NULL AND status <> 'archived'", ) .bind(now) .bind(repo_id) .execute(&mut *txn) .await .map_err(AppError::Database)?; ensure_affected(result.rows_affected(), "repo not found or already archived")?; txn.commit().await.map_err(|_| AppError::TxnError)?; Ok(()) } pub async fn repo_unarchive( &self, ctx: &Session, wk_name: &str, repo_name: &str, ) -> Result<(), AppError> { let user_uid = ctx.user().ok_or(AppError::Unauthorized)?; let repo = self.resolve_repo(wk_name, repo_name).await?; let repo_id = repo.id; self.ensure_repo_role_at_least(user_uid, &repo, Role::Owner) .await?; let now = chrono::Utc::now(); let mut txn = self .ctx .db .writer() .begin() .await .map_err(|_| AppError::TxnError)?; sqlx::query("SET LOCAL app.current_user_id = $1") .bind(user_uid) .execute(&mut *txn) .await .map_err(AppError::Database)?; let result = sqlx::query( "UPDATE repo SET status = 'active', archived_at = NULL, updated_at = $1 WHERE id = $2 AND deleted_at IS NULL AND status = 'archived'", ) .bind(now) .bind(repo_id) .execute(&mut *txn) .await .map_err(AppError::Database)?; ensure_affected(result.rows_affected(), "repo not found or not archived")?; txn.commit().await.map_err(|_| AppError::TxnError)?; Ok(()) } pub async fn repo_delete( &self, ctx: &Session, wk_name: &str, repo_name: &str, ) -> Result<(), AppError> { let user_uid = ctx.user().ok_or(AppError::Unauthorized)?; let repo = self.resolve_repo(wk_name, repo_name).await?; let repo_id = repo.id; self.ensure_repo_role_at_least(user_uid, &repo, Role::Owner) .await?; let now = chrono::Utc::now(); let mut txn = self .ctx .db .writer() .begin() .await .map_err(|_| AppError::TxnError)?; sqlx::query("SET LOCAL app.current_user_id = $1") .bind(user_uid) .execute(&mut *txn) .await .map_err(AppError::Database)?; let result = sqlx::query( "UPDATE repo SET deleted_at = $1, status = 'deleted', updated_at = $1 WHERE id = $2 AND deleted_at IS NULL", ) .bind(now) .bind(repo_id) .execute(&mut *txn) .await .map_err(AppError::Database)?; ensure_affected(result.rows_affected(), "repo not found")?; sqlx::query( "UPDATE workspace_stats SET repos_count = GREATEST(repos_count - 1, 0), updated_at = $1 WHERE workspace_id = $2", ) .bind(now) .bind(repo.workspace_id) .execute(&mut *txn) .await .map_err(AppError::Database)?; txn.commit().await.map_err(|_| AppError::TxnError)?; Ok(()) } pub async fn repo_transfer_owner( &self, ctx: &Session, wk_name: &str, repo_name: &str, new_owner_id: Uuid, ) -> Result { let user_uid = ctx.user().ok_or(AppError::Unauthorized)?; let repo = self.resolve_repo(wk_name, repo_name).await?; let repo_id = repo.id; self.ensure_repo_role_at_least(user_uid, &repo, Role::Owner) .await?; if new_owner_id == repo.owner_id { return Err(AppError::BadRequest( "new owner must be different from current owner".into(), )); } let is_workspace_member = sqlx::query_scalar::<_, bool>( "SELECT EXISTS(SELECT 1 FROM workspace_member WHERE workspace_id = $1 AND user_id = $2 AND status = 'active')", ) .bind(repo.workspace_id) .bind(new_owner_id) .fetch_one(self.ctx.db.reader()) .await .map_err(AppError::Database)?; let is_member = sqlx::query_scalar::<_, bool>( "SELECT EXISTS(SELECT 1 FROM repo_member WHERE repo_id = $1 AND user_id = $2 AND status = 'active')", ) .bind(repo_id) .bind(new_owner_id) .fetch_one(self.ctx.db.reader()) .await .map_err(AppError::Database)?; if !is_workspace_member || !is_member { return Err(AppError::BadRequest( "new owner must be an active workspace and repo member".into(), )); } let now = chrono::Utc::now(); let mut txn = self .ctx .db .writer() .begin() .await .map_err(|_| AppError::TxnError)?; sqlx::query("SET LOCAL app.current_user_id = $1") .bind(user_uid) .execute(&mut *txn) .await .map_err(AppError::Database)?; sqlx::query("UPDATE repo_member SET role = 'owner', updated_at = $1 WHERE repo_id = $2 AND user_id = $3") .bind(now) .bind(repo_id) .bind(new_owner_id) .execute(&mut *txn) .await .map_err(AppError::Database)?; sqlx::query("UPDATE repo_member SET role = 'admin', updated_at = $1 WHERE repo_id = $2 AND user_id = $3") .bind(now) .bind(repo_id) .bind(user_uid) .execute(&mut *txn) .await .map_err(AppError::Database)?; let result = sqlx::query_as::<_, Repo>( "UPDATE repo SET owner_id = $1, updated_at = $2 WHERE id = $3 AND deleted_at IS NULL \ RETURNING id, workspace_id, owner_id, name, description, default_branch, visibility, status, is_fork, forked_from_repo_id, storage_node_ids, primary_storage_node_id, storage_path, git_service, archived_at, created_at, updated_at, deleted_at", ) .bind(new_owner_id) .bind(now) .bind(repo_id) .fetch_one(&mut *txn) .await .map_err(AppError::Database)?; txn.commit().await.map_err(|_| AppError::TxnError)?; Ok(result) } pub(crate) async fn resolve_workspace( &self, wk_name: &str, ) -> Result { crate::models::workspaces::Workspace::find_by_name(self.ctx.db.reader(), wk_name) .await .map_err(AppError::Database)? .ok_or(AppError::NotFound("workspace not found".into())) } pub(crate) async fn resolve_repo( &self, wk_name: &str, repo_name: &str, ) -> Result { let ws = self.resolve_workspace(wk_name).await?; Repo::find_by_name(self.ctx.db.reader(), ws.id, repo_name) .await .map_err(AppError::Database)? .ok_or(AppError::NotFound("repo not found".into())) } pub(crate) async fn find_repo_by_id(&self, repo_id: Uuid) -> Result { sqlx::query_as::<_, Repo>( "SELECT id, workspace_id, owner_id, name, description, default_branch, visibility, status, is_fork, forked_from_repo_id, storage_node_ids, primary_storage_node_id, storage_path, git_service, archived_at, created_at, updated_at, deleted_at FROM repo WHERE id = $1 AND deleted_at IS NULL", ) .bind(repo_id) .fetch_optional(self.ctx.db.reader()) .await .map_err(AppError::Database)? .ok_or(AppError::NotFound("repo not found".into())) } pub async fn repo_user_role( &self, user_uid: Uuid, repo_id: Uuid, ) -> Result, AppError> { let role_str: Option = sqlx::query_scalar( "SELECT role FROM repo_member WHERE repo_id = $1 AND user_id = $2 AND status = 'active'", ) .bind(repo_id) .bind(user_uid) .fetch_optional(self.ctx.db.reader()) .await .map_err(AppError::Database)?; match role_str { Some(r) => Ok(Some(r.parse().unwrap_or(Role::Unknown))), None => { let repo = self.find_repo_by_id(repo_id).await?; if repo.owner_id == user_uid { return Ok(Some(Role::Owner)); } Ok(None) } } } pub async fn ensure_repo_readable(&self, user_uid: Uuid, repo: &Repo) -> Result<(), AppError> { if repo.owner_id == user_uid { return Ok(()); } let is_workspace_member = sqlx::query_scalar::<_, bool>( "SELECT EXISTS(SELECT 1 FROM workspace_member WHERE workspace_id = $1 AND user_id = $2 AND status = 'active')", ) .bind(repo.workspace_id) .bind(user_uid) .fetch_one(self.ctx.db.reader()) .await .map_err(AppError::Database)?; let is_member = sqlx::query_scalar::<_, bool>( "SELECT EXISTS(SELECT 1 FROM repo_member WHERE repo_id = $1 AND user_id = $2 AND status = 'active')", ) .bind(repo.id) .bind(user_uid) .fetch_one(self.ctx.db.reader()) .await .map_err(AppError::Database)?; match repo.visibility { Visibility::Public => Ok(()), Visibility::Internal if is_workspace_member => Ok(()), Visibility::Private if is_workspace_member && is_member => Ok(()), _ => Err(AppError::Unauthorized), } } pub async fn ensure_repo_role_at_least( &self, user_uid: Uuid, repo: &Repo, min_role: Role, ) -> Result { if repo.owner_id == user_uid { return Ok(Role::Owner); } let is_workspace_member = sqlx::query_scalar::<_, bool>( "SELECT EXISTS(SELECT 1 FROM workspace_member WHERE workspace_id = $1 AND user_id = $2 AND status = 'active')", ) .bind(repo.workspace_id) .bind(user_uid) .fetch_one(self.ctx.db.reader()) .await .map_err(AppError::Database)?; if !is_workspace_member { return Err(AppError::Unauthorized); } let role_str: Option = sqlx::query_scalar( "SELECT role FROM repo_member WHERE repo_id = $1 AND user_id = $2 AND status = 'active'", ) .bind(repo.id) .bind(user_uid) .fetch_optional(self.ctx.db.reader()) .await .map_err(AppError::Database)?; let role = role_str .and_then(|r| r.parse::().ok()) .unwrap_or(Role::Unknown); if super::util::role_level(role) < super::util::role_level(min_role) { return Err(AppError::Unauthorized); } Ok(role) } pub(crate) async fn ensure_workspace_readable( &self, user_uid: Uuid, ws: &crate::models::workspaces::Workspace, ) -> Result<(), AppError> { let readable = crate::models::workspaces::Workspace::is_readable(self.ctx.db.reader(), ws, user_uid) .await .map_err(AppError::Database)?; if readable { Ok(()) } else { Err(AppError::Unauthorized) } } pub(crate) async fn ensure_workspace_role_at_least( &self, user_uid: Uuid, ws: &crate::models::workspaces::Workspace, min_role: Role, ) -> Result { let role = crate::models::workspaces::Workspace::user_role( self.ctx.db.reader(), ws.id, user_uid, ws.owner_id, ) .await .map_err(AppError::Database)? .unwrap_or(Role::Unknown); if crate::service::util::role_level(role) < crate::service::util::role_level(min_role) { return Err(AppError::Unauthorized); } Ok(role) } }