"""Projects domain - business logic.""" from __future__ import annotations import re from typing import Any, Optional from uuid import UUID import structlog from sqlalchemy.exc import IntegrityError from src.core.exceptions import ForbiddenError, NotFoundError, PlanLimitError, ValidationError logger = structlog.get_logger() from src.domain.billing.plan_quota_service import PlanQuotaService from src.domain.projects.models import ALLOWED_PROJECT_TYPES, MVP_PROJECT_TYPES, Project from src.domain.projects.schemas import ProjectStackSelectionMode from src.domain.projects.repository import ( ProjectEnvVarRepository, ProjectJobOutboxRepository, ProjectRepository, ProjectRepoRepository, ProjectSettingsRepository, ) from src.domain.projects.interfaces import WorkspaceFacade from src.domain.projects.template_matcher import ( TEMPLATE_MATCH_CONFIDENCE_THRESHOLD, TemplateMatchCandidate, match_template_with_ai, ) from src.domain.templates.models import Template from src.domain.templates.repository import TemplateRepository def slugify(name: str) -> str: """Convert name to URL-safe slug.""" s = name.lower().strip() s = re.sub(r"[^\w\s-]", "", s) s = re.sub(r"[-\s]+", "-", s) return s[:100] or "project" _SLUG_INSERT_MAX_ATTEMPTS = 25 _AUTO_DEFAULT_PROJECT_TYPE = "nuxt_app" _TEMPLATE_AI_MATCH_POOL_SIZE = 12 _TEMPLATE_MATCH_STOPWORDS = { "a", "an", "and", "app", "application", "build", "for", "in", "internal", "of", "or", "page", "pages", "platform", "site", "the", "to", "web", "with", } _INTENT_TYPE_HINTS: tuple[tuple[str, tuple[str, ...]], ...] = ( ("ai_assist_wordpress", ("wordpress", "wp plugin", "blog post", "seo content", "theme change", "plugin update")), ("ai_assist_shopify", ("shopify", "store app", "product sync", "shop assistant", "storefront assistant")), ("html", ("landing page", "marketing site", "brochure", "static site", "blog", "content site", "seo page")), # Product shape is framework-neutral: Nuxt can serve dashboards, portals, # SaaS apps, and admin tools just as well as React. Only infer React from # an explicit React ecosystem signal; otherwise the auto fallback remains Nuxt. ("react_app", ("react", "jsx", "vite")), ) _UNSUPPORTED_MOBILE_HINTS = ("mobile app", "ios app", "android app", "react native", "flutter", "apk") def _is_workspace_slug_conflict(exc: IntegrityError) -> bool: orig = getattr(exc, "orig", None) if orig is None: return False diag = getattr(orig, "diag", None) if diag is not None: cname = getattr(diag, "constraint_name", None) if cname == "uq_projects_workspace_slug": return True msg = str(orig).lower() return "uq_projects_workspace_slug" in msg or ("slug" in msg and "unique" in msg) class ProjectService: """Project business logic.""" def __init__( self, project_repo: ProjectRepository, project_repo_repo: ProjectRepoRepository, settings_repo: ProjectSettingsRepository, env_var_repo: ProjectEnvVarRepository, workspace_facade: WorkspaceFacade, plan_quota: PlanQuotaService, job_outbox_repo: ProjectJobOutboxRepository, template_repo: Optional[TemplateRepository] = None, github_installation_repo: Optional[Any] = None, github_mirror_repo: Optional[Any] = None, gitlab_connection_repo: Optional[Any] = None, gitlab_mirror_repo: Optional[Any] = None, bitbucket_connection_repo: Optional[Any] = None, bitbucket_mirror_repo: Optional[Any] = None, azure_connection_repo: Optional[Any] = None, azure_mirror_repo: Optional[Any] = None, ): self._project_repo = project_repo self._project_repo_repo = project_repo_repo self._settings_repo = settings_repo self._env_var_repo = env_var_repo self._workspace_facade = workspace_facade self._plan_quota = plan_quota self._job_outbox_repo = job_outbox_repo self._template_repo = template_repo self._github_installation_repo = github_installation_repo self._github_mirror_repo = github_mirror_repo self._gitlab_connection_repo = gitlab_connection_repo self._gitlab_mirror_repo = gitlab_mirror_repo self._bitbucket_connection_repo = bitbucket_connection_repo self._bitbucket_mirror_repo = bitbucket_mirror_repo self._azure_connection_repo = azure_connection_repo self._azure_mirror_repo = azure_mirror_repo async def recommend_project_creation( self, *, workspace_id: UUID, requested_type: Optional[str], template_id: Optional[UUID], intent: Optional[str], preferred_stack: Optional[str], stack_selection_mode: str, ) -> dict[str, Any]: resolved_type, reason, warnings = await self._resolve_project_type_for_creation( workspace_id=workspace_id, requested_type=requested_type, template_id=template_id, intent=intent, preferred_stack=preferred_stack, stack_selection_mode=stack_selection_mode, ) template = None template_candidates: list[dict[str, Any]] = [] template_selection_reason: Optional[str] = None if template_id or self._template_repo is not None: template, template_candidates, template_selection_reason = await self._resolve_creation_template( workspace_id=workspace_id, project_type=resolved_type, template_id=template_id, intent=intent, preferred_stack=preferred_stack, suggest=True, ) return { "resolved_project_type": resolved_type, "template_id": getattr(template, "id", None), "template_ref": getattr(template, "ref", None), "template_name": getattr(template, "name", None), "template_selection_reason": template_selection_reason, "template_candidates": template_candidates, "stack_selection_mode": stack_selection_mode, "preferred_stack": preferred_stack, "requested_intent": intent, "decision_reason": reason, "warnings": warnings, } async def _resolve_creation_template( self, *, workspace_id: UUID, project_type: str, template_id: Optional[UUID], intent: Optional[str] = None, preferred_stack: Optional[str] = None, suggest: bool = False, ) -> tuple[Optional[Template], list[dict[str, Any]], Optional[str]]: """Resolve template policy. Create binds a GitHub starter only when the caller sent ``template_id``. Recommendation may still suggest a closest/default template for the UI; declining that suggestion must take the generated S2 assembly path. """ if self._template_repo is None: return None, [], None plan_slug = await self._plan_quota.plan_slug(workspace_id) if template_id is not None: template = await self._template_repo.get_by_id(template_id) if not template or not template.is_active: raise ValidationError("Template is not available.") if template.framework != project_type: raise ValidationError("Template does not match the requested project type.") if not await self._template_repo.plan_access_allows(template.id, plan_slug): raise ValidationError("Template is not available for this workspace plan.") return template, [], "User explicitly selected this template." if not suggest: return None, [], None matched, candidates, match_reason = await self._find_matching_templates( plan_slug=plan_slug, project_type=project_type, intent=intent, preferred_stack=preferred_stack, ) if matched is not None: return matched, candidates, match_reason or "Auto-selected the closest existing template for this intent." default_template = await self._template_repo.get_default_for_framework(project_type, plan_slug) reason = "Fell back to the framework default template." return default_template, candidates, reason if default_template is not None else None async def create_project( self, workspace_id: UUID, name: str, project_type: Optional[str], description: Optional[str], user_id: UUID, template_id: Optional[UUID] = None, source: Optional[dict] = None, project_id: Optional[UUID] = None, intent: Optional[str] = None, preferred_stack: Optional[str] = None, stack_selection_mode: str = ProjectStackSelectionMode.AUTO.value, discovery_requested: bool = False, skip_discovery: bool = False, ) -> Project: """Create project. Returns 202 - repo creation is async. skip_discovery is for callers that already have a full user_prompt to give start_build directly (agent-created projects via MCP's create_project -- see AgentProjectFacade.create) -- the interactive discovery Q&A this seeds is never going to be walked/completed for those, and require_discovery_complete would otherwise block every future start_build on the project forever, since nothing else ever marks discovery "completed" outside the Dashboard's own UI. """ if not await self._workspace_facade.check_permission(workspace_id, user_id, "projects.create"): raise ForbiddenError() if source and not await self._workspace_facade.check_permission(workspace_id, user_id, "repos.manage"): raise ForbiddenError() from src.config import get_settings from src.domain.projects.discovery_service import discovery_enabled_for_workspace discovery_settings = get_settings() if discovery_requested and not discovery_enabled_for_workspace(discovery_settings, workspace_id): raise ValidationError( "Discovery pilot is not enabled for this workspace on this backend. " "Load the pilot configuration before creating a test project." ) discovery_enabled = ( not skip_discovery and discovery_enabled_for_workspace(discovery_settings, workspace_id) and (discovery_settings.project_discovery_public_enabled or discovery_requested) ) project_type, _reason, _warnings = await self._resolve_project_type_for_creation( workspace_id=workspace_id, requested_type=project_type, template_id=template_id, intent=intent, preferred_stack=preferred_stack, stack_selection_mode=stack_selection_mode, ) if project_type not in ALLOWED_PROJECT_TYPES: raise ValueError(f"Invalid project type. Allowed: {ALLOWED_PROJECT_TYPES}") if template_id and source: raise ValidationError("Template creation cannot be combined with a source repo import.") source_info = await self._validate_import_source(workspace_id, source) if source else None # Keep project creation serialized for slug, quota, and initial-job consistency. await self._project_repo.acquire_workspace_advisory_xact_lock(workspace_id) maximum = await self._plan_quota.max_projects(workspace_id) if maximum is not None: existing = await self._project_repo.list_for_workspace(workspace_id) if len(existing) >= maximum: raise PlanLimitError( f"Workspace project limit reached ({maximum}).", limit_code="max_projects", current_count=len(existing), max_allowed=maximum, ) template = None if source_info is None: template, _template_candidates, _template_selection_reason = await self._resolve_creation_template( workspace_id=workspace_id, project_type=project_type, template_id=template_id, intent=intent, preferred_stack=preferred_stack, suggest=False, ) base_slug = slugify(name) slug_candidate = base_slug counter = 0 project: Optional[Project] = None for _ in range(_SLUG_INSERT_MAX_ATTEMPTS): while await self._project_repo.slug_exists_in_workspace(workspace_id, slug_candidate): counter += 1 slug_candidate = f"{base_slug}-{counter}" try: async with self._project_repo.session.begin_nested(): project = await self._project_repo.create( workspace_id=workspace_id, slug=slug_candidate, name=name, project_type=project_type, description=description, created_by=user_id, template_id=template.id if template else None, template_ref=template.ref if template else None, template_github_repo=template.github_repo if template else None, template_starter_branch=template.starter_branch if template else None, template_starter_sha=template.starter_sha if template else None, source_kind=source_info["kind"] if source_info else "generated", source_workspace_user_github_installation_id=( source_info.get("installation_row_id") if source_info else None ), source_workspace_user_gitlab_connection_id=( source_info.get("connection_row_id") if source_info else None ), source_workspace_user_bitbucket_connection_id=( source_info.get("bitbucket_connection_row_id") if source_info else None ), source_workspace_user_azure_connection_id=( source_info.get("azure_connection_row_id") if source_info else None ), source_repo_full_name=source_info["repo_full_name"] if source_info else None, source_branch=source_info["branch"] if source_info else None, source_root_dir=source_info["root_dir"] if source_info else None, source_import_env=source_info["import_env"] if source_info else True, project_id=project_id, ) except IntegrityError as exc: if _is_workspace_slug_conflict(exc): counter += 1 slug_candidate = f"{base_slug}-{counter}" continue raise break if project is None: raise RuntimeError("Could not allocate project slug after uniqueness retries") await self._settings_repo.get_or_create(project.id) if source_info and self._github_mirror_repo is not None: if source_info["kind"] == "github_import": await self._github_mirror_repo.upsert( project_id=project.id, workspace_user_github_installation_id=source_info["installation_row_id"], target_repo_full_name=source_info["repo_full_name"], target_default_branch=source_info["branch"], enabled=True, ) if source_info and self._gitlab_mirror_repo is not None: if source_info["kind"] == "gitlab_import": await self._gitlab_mirror_repo.upsert( project_id=project.id, workspace_user_gitlab_connection_id=source_info["connection_row_id"], target_project_full_path=source_info["repo_full_name"], target_default_branch=source_info["branch"], enabled=True, ) if source_info and self._bitbucket_mirror_repo is not None: if source_info["kind"] == "bitbucket_import": await self._bitbucket_mirror_repo.upsert( project_id=project.id, workspace_user_bitbucket_connection_id=source_info[ "bitbucket_connection_row_id" ], target_repo_full_name=source_info["repo_full_name"], target_default_branch=source_info["branch"], enabled=True, ) if source_info and self._azure_mirror_repo is not None: if source_info["kind"] == "azure_devops_import": await self._azure_mirror_repo.upsert( project_id=project.id, workspace_user_azure_connection_id=source_info["azure_connection_row_id"], target_repo_full_name=source_info["repo_full_name"], target_default_branch=source_info["branch"], enabled=True, ) if discovery_enabled: from src.domain.projects.discovery_service import initial_discovery project.discovery_json = initial_discovery( intent or "", locked_type=(project_type if project_type and ( template_id or source_info or stack_selection_mode == "require" ) else None), ) # Creation/provisioning is unchanged; discovery gates the first build only. await self._project_repo.session.flush() await self._job_outbox_repo.insert_initial_project_jobs(project.id) # Best-effort latency nudge (cron still publishes outbox rows). Must not fail creation if Redis is down. from src.workers.enqueue import enqueue_project_outbox_dispatch_nudge try: await enqueue_project_outbox_dispatch_nudge(defer_seconds=3.0) except Exception: logger.warning("project_outbox_dispatch_nudge_failed", exc_info=True) return project async def _resolve_project_type_for_creation( self, *, workspace_id: UUID, requested_type: Optional[str], template_id: Optional[UUID], intent: Optional[str], preferred_stack: Optional[str], stack_selection_mode: str, ) -> tuple[str, str, list[str]]: requested = (requested_type or "").strip() preferred = (preferred_stack or "").strip() mode = (stack_selection_mode or ProjectStackSelectionMode.AUTO.value).strip().lower() intent_text = (intent or "").strip() warnings: list[str] = [] inferred_from_intent = self._infer_project_type_from_intent(intent_text) if inferred_from_intent == "unsupported_mobile": raise ValidationError( "Mobile app requests are not supported by this project creation flow yet. " "Use a supported web/plugin target or add a dedicated mobile project type first." ) if template_id is not None: if self._template_repo is None: raise ValidationError("Template selection is not configured.") template = await self._template_repo.get_by_id(template_id) if not template or not template.is_active: raise ValidationError("Template is not available.") template_framework = (template.framework or "").strip() if requested and requested != template_framework: raise ValidationError("Template does not match the requested project type.") if preferred and mode == ProjectStackSelectionMode.REQUIRE.value and preferred != template_framework: raise ValidationError("Template does not match the required preferred stack.") return template_framework, "Resolved from selected template.", warnings if requested: if preferred and mode == ProjectStackSelectionMode.REQUIRE.value and preferred != requested: raise ValidationError("Requested project type does not match the required preferred stack.") return requested, "Resolved from explicit project type.", warnings if preferred: if inferred_from_intent and inferred_from_intent != preferred and mode == ProjectStackSelectionMode.REQUIRE.value: warnings.append( f"Required stack override '{preferred}' takes precedence over inferred intent '{inferred_from_intent}'." ) if mode in ( ProjectStackSelectionMode.PREFER.value, ProjectStackSelectionMode.REQUIRE.value, ): reason = ( "Resolved from required power-user stack override." if mode == ProjectStackSelectionMode.REQUIRE.value else "Resolved from preferred power-user stack override." ) return preferred, reason, warnings if inferred_from_intent: return inferred_from_intent, "Resolved automatically from project intent.", warnings fallback = preferred or _AUTO_DEFAULT_PROJECT_TYPE return fallback, "Resolved from default auto-selection fallback.", warnings async def _find_matching_templates( self, *, plan_slug: str, project_type: str, intent: Optional[str], preferred_stack: Optional[str], limit: int = 3, ) -> tuple[Optional[Template], list[dict[str, Any]], Optional[str]]: if self._template_repo is None: return None, [], None intent_text = (intent or "").strip() if not intent_text: return None, [], None templates = await self._template_repo.list_for_plan(plan_slug) same_type = [t for t in templates if (t.framework or "").strip() == project_type] if not same_type: return None, [], None # Score every same-framework template (not just intent.keyword hits) — a semantic match # can have zero literal term overlap (e.g. "handmade jewelry shop" vs. tags "ecommerce, # boutique"), and the AI matcher below needs the full ranked pool to catch those. scored: list[tuple[int, list[str], Template]] = [ ( *self._score_template_match(template=t, intent_text=intent_text.lower(), preferred_stack=preferred_stack), t, ) for t in same_type ] scored.sort( key=lambda item: ( item[0], 1 if getattr(item[2], "is_default", False) else 0, str(getattr(item[2], "updated_at", "") or ""), str(getattr(item[2], "created_at", "") or ""), ), reverse=True, ) keyword_matched = [item for item in scored if item[0] > 0] candidates = [ { "template_id": template.id, "template_ref": template.ref, "template_name": template.name, "framework": template.framework, "score": score, "matched_terms": matched_terms, } for score, matched_terms, template in keyword_matched[:limit] ] ai_pool = scored[:_TEMPLATE_AI_MATCH_POOL_SIZE] ai_result = await match_template_with_ai( intent_text, [ TemplateMatchCandidate( ref=template.ref, name=template.name, description=template.description, tags=list(getattr(template, "tags", None) or []), ) for _, _, template in ai_pool ], ) if ai_result is not None and ai_result.confidence >= TEMPLATE_MATCH_CONFIDENCE_THRESHOLD: matched_template = next((t for _, _, t in ai_pool if t.ref == ai_result.template_ref), None) if matched_template is not None: return matched_template, candidates, ai_result.reasoning or None from src.config import get_settings if get_settings().project_intake_enabled: # An explicit semantic "no good base" must not become an unrelated keyword match. return None, candidates, None if not keyword_matched: return None, candidates, None return keyword_matched[0][2], candidates, None def _score_template_match( self, *, template: Template, intent_text: str, preferred_stack: Optional[str], ) -> tuple[int, list[str]]: template_terms = self._template_search_terms(template) intent_terms = self._keyword_terms(intent_text) if not intent_terms: return 0, [] matched_terms = sorted(intent_terms & template_terms) score = len(matched_terms) * 10 for tag in self._template_tag_phrases(template): if tag and tag in intent_text: score += 12 matched_terms.append(tag) if preferred_stack and preferred_stack == getattr(template, "framework", None): score += 3 unique_terms = sorted({term for term in matched_terms if term}) return score, unique_terms def _template_search_terms(self, template: Template) -> set[str]: parts = [ getattr(template, "name", "") or "", getattr(template, "ref", "") or "", getattr(template, "description", "") or "", " ".join(self._template_tag_phrases(template)), ] return self._keyword_terms(" ".join(parts)) def _template_tag_phrases(self, template: Template) -> list[str]: raw_tags = getattr(template, "tags", None) or [] phrases: list[str] = [] for tag in raw_tags: text = str(tag or "").strip().lower() if text: phrases.append(text) return phrases def _keyword_terms(self, text: str) -> set[str]: terms = { token for token in re.findall(r"[a-z0-9]+", (text or "").lower()) if len(token) >= 3 and token not in _TEMPLATE_MATCH_STOPWORDS } return terms def _infer_project_type_from_intent(self, intent: str) -> Optional[str]: text = (intent or "").strip().lower() if not text: return None if any(hint in text for hint in _UNSUPPORTED_MOBILE_HINTS): return "unsupported_mobile" for project_type, hints in _INTENT_TYPE_HINTS: if any(hint in text for hint in hints): return project_type return None async def _validate_import_source(self, workspace_id: UUID, source: Optional[dict]) -> Optional[dict]: if not source: return None kind = (source.get("kind") or "github_import").strip() if kind == "gitlab_import": return await self._validate_gitlab_import_source(workspace_id, source) if kind == "bitbucket_import": return await self._validate_bitbucket_import_source(workspace_id, source) if kind == "azure_devops_import": return await self._validate_azure_import_source(workspace_id, source) return await self._validate_github_import_source(workspace_id, source) async def _validate_github_import_source(self, workspace_id: UUID, source: Optional[dict]) -> Optional[dict]: if not source: return None if self._github_installation_repo is None: raise ValidationError("GitHub import is not configured.") installation_row_id = source.get("workspace_user_github_installation_id") repo_full_name = (source.get("repo_full_name") or "").strip() requested_branch = (source.get("branch") or "").strip() root_dir = (source.get("root_dir") or "/").strip().replace("\\", "/") root_dir = "/" if root_dir in ("", ".") else root_dir.strip("/") or "/" if not installation_row_id or not repo_full_name: raise ValidationError("GitHub import source requires installation and repo.") inst = await self._github_installation_repo.get_by_id(installation_row_id) if not inst or inst.workspace_id != workspace_id or inst.suspended_at is not None: raise ValidationError("Invalid or suspended GitHub installation for this workspace.") from src.integrations.github_user_app_client import GitHubUserAppClient gh = GitHubUserAppClient() repos = await gh.list_installation_repositories(inst.github_installation_id) visible = { str(r.get("full_name") or ""): str(r.get("default_branch") or "main") for r in repos if r.get("full_name") } if repo_full_name not in visible: raise ValidationError("Source repo is not visible to this installation.") return { "kind": "github_import", "installation_row_id": installation_row_id, "github_installation_id": inst.github_installation_id, "repo_full_name": repo_full_name, "branch": requested_branch or visible[repo_full_name] or "main", "root_dir": root_dir, "import_env": bool(source.get("import_env", True)), } async def _validate_gitlab_import_source(self, workspace_id: UUID, source: dict) -> dict: if self._gitlab_connection_repo is None: raise ValidationError("GitLab import is not configured.") connection_row_id = source.get("workspace_user_gitlab_connection_id") repo_full_name = (source.get("repo_full_name") or "").strip() requested_branch = (source.get("branch") or "").strip() root_dir = (source.get("root_dir") or "/").strip().replace("\\", "/") root_dir = "/" if root_dir in ("", ".") else root_dir.strip("/") or "/" if not connection_row_id or not repo_full_name: raise ValidationError("GitLab import source requires connection and project.") conn = await self._gitlab_connection_repo.get_by_id(connection_row_id) if not conn or conn.workspace_id != workspace_id or conn.revoked_at is not None: raise ValidationError("Invalid or revoked GitLab connection for this workspace.") from src.core.security.encryption import decrypt_value from src.integrations.gitlab_user_client import GitLabUserClient client = GitLabUserClient(access_token=decrypt_value(conn.access_token_ciphertext)) project = await client.get_project_by_path(repo_full_name) default_branch = str(project.get("default_branch") or "main") return { "kind": "gitlab_import", "connection_row_id": connection_row_id, "repo_full_name": repo_full_name, "branch": requested_branch or default_branch, "root_dir": root_dir, "import_env": bool(source.get("import_env", True)), } async def _validate_bitbucket_import_source(self, workspace_id: UUID, source: dict) -> dict: if self._bitbucket_connection_repo is None: raise ValidationError("Bitbucket import is not configured.") connection_row_id = source.get("workspace_user_bitbucket_connection_id") repo_full_name = (source.get("repo_full_name") or "").strip() requested_branch = (source.get("branch") or "").strip() root_dir = (source.get("root_dir") or "/").strip().replace("\\", "/") root_dir = "/" if root_dir in ("", ".") else root_dir.strip("/") or "/" if not connection_row_id or not repo_full_name: raise ValidationError("Bitbucket import source requires connection and repo.") conn = await self._bitbucket_connection_repo.get_by_id(connection_row_id) if not conn or conn.workspace_id != workspace_id or conn.revoked_at is not None: raise ValidationError("Invalid or revoked Bitbucket connection for this workspace.") from src.core.security.encryption import decrypt_value from src.integrations.bitbucket_user_client import BitbucketUserClient client = BitbucketUserClient(access_token=decrypt_value(conn.access_token_ciphertext)) repo = await client.get_repository(repo_full_name) default_branch = str((repo.get("mainbranch") or {}).get("name") or "main") return { "kind": "bitbucket_import", "bitbucket_connection_row_id": connection_row_id, "repo_full_name": repo_full_name, "branch": requested_branch or default_branch, "root_dir": root_dir, "import_env": bool(source.get("import_env", True)), } async def _validate_azure_import_source(self, workspace_id: UUID, source: dict) -> dict: if self._azure_connection_repo is None: raise ValidationError("Azure DevOps import is not configured.") connection_row_id = source.get("workspace_user_azure_connection_id") repo_full_name = (source.get("repo_full_name") or "").strip() requested_branch = (source.get("branch") or "").strip() root_dir = (source.get("root_dir") or "/").strip().replace("\\", "/") root_dir = "/" if root_dir in ("", ".") else root_dir.strip("/") or "/" if not connection_row_id or not repo_full_name: raise ValidationError("Azure DevOps import source requires connection and repo.") conn = await self._azure_connection_repo.get_by_id(connection_row_id) if not conn or conn.workspace_id != workspace_id or conn.revoked_at is not None: raise ValidationError("Invalid or revoked Azure DevOps connection for this workspace.") from src.domain.azure_user_mirror.service import resolve_azure_connection_client from src.integrations.azure_devops_client import split_azure_repo_full_name azure_project, azure_repo = split_azure_repo_full_name(repo_full_name) client = await resolve_azure_connection_client(conn, self._azure_connection_repo) repo = await client.get_repository(azure_project, azure_repo) default_branch = client.default_branch_name(repo) return { "kind": "azure_devops_import", "azure_connection_row_id": connection_row_id, "repo_full_name": repo_full_name, "branch": requested_branch or default_branch, "root_dir": root_dir, "import_env": bool(source.get("import_env", True)), } async def get_project(self, project_id: UUID, user_id: UUID) -> Optional[Project]: project = await self._project_repo.get_with_repo(project_id) if not project: return None if not await self._workspace_facade.check_permission(project.workspace_id, user_id, "projects.read"): return None return project async def list_projects(self, workspace_id: UUID, user_id: UUID) -> list[Project]: if not await self._workspace_facade.check_permission(workspace_id, user_id, "projects.read"): raise ForbiddenError() return await self._project_repo.list_for_workspace(workspace_id) async def update_project( self, project_id: UUID, user_id: UUID, name: Optional[str] = None, description: Optional[str] = None ) -> Optional[Project]: project = await self._project_repo.get_by_id(project_id) if not project: return None if not await self._workspace_facade.check_permission(project.workspace_id, user_id, "projects.write"): raise ForbiddenError() await self._project_repo.update(project_id, name=name, description=description) return await self._project_repo.get_with_repo(project_id) async def delete_project(self, project_id: UUID, user_id: UUID) -> bool: project = await self._project_repo.get_by_id(project_id) if not project: return False if not await self._workspace_facade.check_permission(project.workspace_id, user_id, "projects.delete"): raise ForbiddenError() await self._project_repo.soft_delete(project_id) return True async def add_env_var( self, project_id: UUID, user_id: UUID, key: str, value: str, environment: str = "all", visibility: str = "secret", ): """Add encrypted env var.""" from src.core.security.encryption import encrypt_value project = await self._project_repo.get_by_id(project_id) if not project: raise NotFoundError("Project not found") if not await self._workspace_facade.check_permission(project.workspace_id, user_id, "secrets.manage"): raise ForbiddenError() return await self._env_var_repo.create( project_id=project_id, key=key, encrypted_value=encrypt_value(value), environment=environment, visibility=visibility, ) async def list_env_vars( self, project_id: UUID, user_id: UUID, environment: Optional[str] = None, ): project = await self._project_repo.get_by_id(project_id) if not project: raise NotFoundError("Project not found") if not await self._workspace_facade.check_permission(project.workspace_id, user_id, "secrets.manage"): raise ForbiddenError() return await self._env_var_repo.list_for_project(project_id, environment=environment) async def update_env_var( self, project_id: UUID, env_var_id: UUID, user_id: UUID, value: str, visibility: Optional[str] = None, ): from src.core.security.encryption import encrypt_value project = await self._project_repo.get_by_id(project_id) if not project: raise NotFoundError("Project not found") if not await self._workspace_facade.check_permission(project.workspace_id, user_id, "secrets.manage"): raise ForbiddenError() ev = await self._env_var_repo.get_by_id(project_id, env_var_id) if not ev: raise NotFoundError("Environment variable not found") return await self._env_var_repo.update_encrypted_value( env_var_id, encrypt_value(value), visibility=visibility ) async def delete_env_var( self, project_id: UUID, env_var_id: UUID, user_id: UUID ) -> bool: project = await self._project_repo.get_by_id(project_id) if not project: raise NotFoundError("Project not found") if not await self._workspace_facade.check_permission(project.workspace_id, user_id, "secrets.manage"): raise ForbiddenError() deleted = await self._env_var_repo.delete_by_id(project_id, env_var_id) if not deleted: raise NotFoundError("Environment variable not found") return True async def get_env_vars_for_propagation(self, project_id: UUID) -> list[dict]: """Return decrypted env var entries for the propagation adapter. Internal use only.""" from src.core.security.encryption import decrypt_value rows = await self._env_var_repo.list_all_for_deploy(project_id) return [ { "key": r.key, "value": decrypt_value(r.encrypted_value), "visibility": r.visibility, "environment": r.environment, } for r in rows ]