ss-tools/backend/src/api/routes/git/_repo_lifecycle_routes.py

# #region GitRepoLifecycleRoutes [C:3] [TYPE Module] [SEMANTICS fastapi, git, api, sync, deploy]
# @defgroup Api Module group.
# @BRIEF FastAPI endpoints for Git lifecycle operations (sync, promote, deploy).
# @LAYER API


from pathlib import Path
from datetime import UTC, datetime
import io
import tempfile
import zipfile

from fastapi import Depends, HTTPException
from sqlalchemy.orm import Session

from src.api.routes.git_schemas import (
    DeployRequest,
    DeploymentValidationRequest,
    DeploymentStatusResponse,
    EnvironmentDeploymentStatus,
    PromoteRequest,
    PromoteResponse,
)
from src.core.database import get_db
from src.core.logger import belief_scope, logger
from src.core.superset_client import SupersetClient
from src.dependencies import get_config_manager, get_current_user, has_permission
from src.models.auth import User
from src.models.git import GitProvider, GitRepository

from ._deps import get_git_service
from ._helpers import (
    _apply_git_identity_from_profile,
    _get_git_config_or_404,
    _handle_unexpected_git_route_error,
)
from ._router import router


_STAGE_CANONICAL: dict[str, str] = {
    "dev": "dev",
    "development": "dev",
    "разработка": "dev",
    "preprod": "preprod",
    "pre-production": "preprod",
    "staging": "preprod",
    "препрод": "preprod",
    "prod": "prod",
    "production": "prod",
    "продакшн": "prod",
    "прод": "prod",
}


def _canonicalize_stage(name: str) -> str:
    """Normalize environment name to canonical stage key (dev/preprod/prod).

    Edge B3: Maps DeploymentEnvironment name (e.g. 'Production') to canonical key
    understood by frontend timeline ENVIRONMENTS array ['dev', 'preprod', 'prod'].
    """
    key = name.strip().lower().replace(" ", "_").replace("-", "_")
    if key in _STAGE_CANONICAL:
        return _STAGE_CANONICAL[key]
    if "preprod" in key or "pre_production" in key:
        return "preprod"
    if "prod" in key:
        return "prod"
    if "dev" in key or "development" in key:
        return "dev"
    return key


# #region GitDeployment.resolve_stage_environment [C:3] [TYPE Function] [SEMANTICS git,deployment,stage]
# @ingroup Api
# @BRIEF Resolve a canonical release stage to one internal deployment connection.
# @PRE stage is one of the public lifecycle stages; only PREPROD and PROD are deploy targets.
# @POST Returns exactly one active DeploymentEnvironment or raises a descriptive HTTP error.
# @RATIONALE The HTTP contract names a lifecycle stage, while server connection IDs remain internal infrastructure.
# @REJECTED Accepting environment IDs from the client was rejected because UI semantics then leak connection topology.
def _resolve_stage_environment(stage: str, db: Session, config_manager):
    from src.models.git import DeploymentEnvironment

    canonical_stage = _canonicalize_stage(stage)
    configured = [
        environment
        for environment in config_manager.get_environments()
        if _canonicalize_stage(str(environment.stage)) == canonical_stage
    ]
    if len(configured) != 1:
        raise HTTPException(status_code=404, detail=f"No unique active environment is configured for stage '{canonical_stage}'")

    source = configured[0]
    persisted = db.query(DeploymentEnvironment).filter(DeploymentEnvironment.id == str(source.id)).first()
    if persisted:
        return persisted
    target = DeploymentEnvironment(
        id=str(source.id),
        name=str(source.name),
        superset_url=str(source.url),
        superset_token="",
        is_active=True,
    )
    db.add(target)
    db.commit()
    return target


# #endregion GitDeployment.resolve_stage_environment


# #region GitDeployment.enforce_approval_policy [C:2] [TYPE Function] [SEMANTICS git,release,approval]
# @ingroup Api
# @BRIEF Enforce the configurable role/comment gate for an immutable release candidate.
def _enforce_approval_policy(policy, user: User, comment: str | None) -> None:
    role_names = {role.name for role in user.roles}
    if policy.approval_roles and not role_names.intersection(policy.approval_roles):
        raise HTTPException(status_code=403, detail="Your role cannot approve dashboard releases")
    if policy.require_approval_comment and not str(comment or "").strip():
        raise HTTPException(status_code=422, detail="Release policy requires an approval comment")


# #endregion GitDeployment.enforce_approval_policy


# #region GitDeployment.probe_drift [C:4] [TYPE Function] [SEMANTICS git,deployment,drift,superset]
# @ingroup Api
# @BRIEF Compare a recorded candidate fingerprint with the dashboard currently exported from its target Superset.
# @POST Returns unknown rather than failing the lifecycle view when the target is unreachable.
# @SIDE_EFFECT Reads a target Superset dashboard export without mutating Git or Superset.
async def _probe_drift(dashboard_ref: str, environment_id: str, expected_hash: str, config_manager) -> tuple[str, str | None]:
    from . import _resolve_dashboard_id_from_ref
    from src.plugins.git_fingerprint import _compute_content_hash

    try:
        environment = config_manager.get_environment(environment_id)
        if not environment:
            return "unknown", None
        dashboard_id = await _resolve_dashboard_id_from_ref(dashboard_ref, config_manager, environment_id)
        client = SupersetClient(environment)
        await client.authenticate()
        archive_bytes, _ = await client.export_dashboard(dashboard_id)
        with tempfile.TemporaryDirectory(prefix="superset-tools-drift-") as directory:
            root = Path(directory)
            with zipfile.ZipFile(io.BytesIO(archive_bytes)) as archive:
                archive.extractall(root)
            metadata = next(root.rglob("metadata.yaml"), None)
            actual_hash = _compute_content_hash(metadata.parent) if metadata else None
        if not actual_hash:
            return "unknown", None
        return ("in_sync" if actual_hash == expected_hash else "drifted"), actual_hash
    except Exception as exc:
        logger.explore("Could not probe deployment drift", error=str(exc), payload={"environment_id": environment_id})
        return "unknown", None


# #endregion GitDeployment.probe_drift


# #region sync_dashboard [C:3] [TYPE Function]
# @ingroup Api
# @BRIEF Sync dashboard state from Superset to Git using the GitPlugin.
# @RELATION CALLS -> [GitPlugin]
@router.post("/repositories/{dashboard_ref}/sync")
async def sync_dashboard(
    dashboard_ref: str,
    env_id: str | None = None,
    source_env_id: str | None = None,
    config_manager=Depends(get_config_manager),
    _=Depends(has_permission("plugin:git", "EXECUTE")),
):
    with belief_scope("sync_dashboard"):
        from . import _resolve_dashboard_id_from_ref

        try:
            dashboard_id = await _resolve_dashboard_id_from_ref(dashboard_ref, config_manager, env_id)
            from src.plugins.git_plugin import GitPlugin

            plugin = GitPlugin()
            result = await plugin.execute(
                {
                    "operation": "sync",
                    "dashboard_id": dashboard_id,
                    "source_env_id": source_env_id,
                }
            )
            return result
        except HTTPException:
            raise
        except Exception as e:
            _handle_unexpected_git_route_error("sync_dashboard", e)


# #endregion sync_dashboard


# #region promote_dashboard [C:3] [TYPE Function]
# @ingroup Api
# @BRIEF Promote changes between branches via MR or direct merge.
# @RELATION CALLS -> [GitPlugin]
@router.post("/repositories/{dashboard_ref}/promote", response_model=PromoteResponse)
async def promote_dashboard(
    dashboard_ref: str,
    payload: PromoteRequest,
    env_id: str | None = None,
    config_manager=Depends(get_config_manager),
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_user),
    _=Depends(has_permission("plugin:git", "EXECUTE")),
):
    _gs = get_git_service()
    with belief_scope("promote_dashboard"):
        from . import _resolve_dashboard_id_from_ref
        from ._helpers import _handle_unexpected_git_route_error

        try:
            dashboard_id = await _resolve_dashboard_id_from_ref(dashboard_ref, config_manager, env_id)
            db_repo = db.query(GitRepository).filter(GitRepository.dashboard_id == dashboard_id).first()
            if not db_repo:
                raise HTTPException(
                    status_code=404,
                    detail=f"Repository for dashboard {dashboard_ref} is not initialized",
                )
            config = _get_git_config_or_404(db, db_repo.config_id)

            from_branch = payload.from_branch.strip()
            to_branch = payload.to_branch.strip()
            if not from_branch or not to_branch:
                raise HTTPException(status_code=400, detail="from_branch and to_branch are required")
            if from_branch == to_branch:
                raise HTTPException(status_code=400, detail="from_branch and to_branch must be different")

            mode = (payload.mode or "mr").strip().lower()
            if mode == "direct":
                reason = (payload.reason or "").strip()
                if not reason:
                    raise HTTPException(status_code=400, detail="Direct promote requires non-empty reason")
                logger.warning(
                    "[promote_dashboard][PolicyViolation] Direct promote without MR by actor=unknown dashboard_ref=%s from=%s to=%s reason=%s",
                    dashboard_ref,
                    from_branch,
                    to_branch,
                    reason,
                )
                await _apply_git_identity_from_profile(dashboard_id, db, current_user)
                result = await _gs.promote_direct_merge(
                    dashboard_id=dashboard_id,
                    from_branch=from_branch,
                    to_branch=to_branch,
                )
                return PromoteResponse(
                    mode="direct",
                    from_branch=from_branch,
                    to_branch=to_branch,
                    status=result.get("status", "merged"),
                    policy_violation=True,
                )

            title = (payload.title or "").strip() or f"Promote {from_branch} -> {to_branch}"
            description = payload.description
            if config.provider == GitProvider.GITEA:
                pr = await _gs.create_gitea_pull_request(
                    server_url=config.url,
                    pat=config.pat,
                    remote_url=db_repo.remote_url,
                    from_branch=from_branch,
                    to_branch=to_branch,
                    title=title,
                    description=description,
                )
            elif config.provider == GitProvider.GITHUB:
                pr = await _gs.create_github_pull_request(
                    server_url=config.url,
                    pat=config.pat,
                    remote_url=db_repo.remote_url,
                    from_branch=from_branch,
                    to_branch=to_branch,
                    title=title,
                    description=description,
                    draft=payload.draft,
                )
            elif config.provider == GitProvider.GITLAB:
                pr = await _gs.create_gitlab_merge_request(
                    server_url=config.url,
                    pat=config.pat,
                    remote_url=db_repo.remote_url,
                    from_branch=from_branch,
                    to_branch=to_branch,
                    title=title,
                    description=description,
                    remove_source_branch=payload.remove_source_branch,
                )
            else:
                raise HTTPException(
                    status_code=501,
                    detail=f"Provider {config.provider} does not support promotion API",
                )

            return PromoteResponse(
                mode="mr",
                from_branch=from_branch,
                to_branch=to_branch,
                status=pr.get("status", "opened"),
                url=pr.get("url"),
                reference_id=str(pr.get("id")) if pr.get("id") is not None else None,
                policy_violation=False,
            )
        except HTTPException:
            raise
        except Exception as e:
            _handle_unexpected_git_route_error("promote_dashboard", e)


# #endregion promote_dashboard


# #region get_deployment_status [C:3] [TYPE Function] [SEMANTICS deployment, versioning, status]
# @ingroup Api
# @BRIEF Get per-environment deployment status with content-hash comparison.
# @RELATION CALLS -> [_handle_deploy_helpers._get_last_deployment]
# @RELATION CALLS -> [_handle_deploy_helpers._compute_content_hash]
@router.get("/repositories/{dashboard_ref}/deployment-status", response_model=DeploymentStatusResponse)
async def get_deployment_status(
    dashboard_ref: str,
    env_id: str | None = None,
    config_manager=Depends(get_config_manager),
    _=Depends(has_permission("plugin:git", "EXECUTE")),
):
    with belief_scope("get_deployment_status"):
        from . import _resolve_dashboard_id_from_ref
        from src.plugins.git_fingerprint import _compute_content_hash
        from src.plugins.git_deployment_recorder import _get_last_deployment

        try:
            dashboard_id = await _resolve_dashboard_id_from_ref(dashboard_ref, config_manager, env_id)
            from src.services.git_service import GitService

            gs = GitService()
            repo = await gs.get_repo(dashboard_id)
            repo_path = Path(repo.working_dir)

            # Current content hash (dev branch)
            current_hash = _compute_content_hash(repo_path)

            # Collect deployment status for each environment
            from src.core.database import SessionLocal
            from src.models.git import DeploymentEnvironment, GitRepository

            db = SessionLocal()
            try:
                # Get repository_id for scoped deployment lookup (FIX A1)
                git_repo = db.query(GitRepository).filter(GitRepository.dashboard_id == dashboard_id).first()
                repository_id = git_repo.id if git_repo else None

                envs = (
                    db.query(DeploymentEnvironment)
                    .filter(
                        DeploymentEnvironment.is_active == True  # noqa: E712
                    )
                    .all()
                )
                configured_stages = {
                    str(environment.id): _canonicalize_stage(str(environment.stage))
                    for environment in config_manager.get_environments()
                }

                environments = []
                for stage in ("dev", "preprod", "prod"):
                    stage_envs = [environment for environment in envs if configured_stages.get(str(environment.id)) == stage]
                    deployments = [
                        _get_last_deployment(repository_id, environment.id, db_session=db)
                        for environment in stage_envs
                    ] if repository_id else []
                    last = max(
                        (deployment for deployment in deployments if deployment),
                        key=lambda deployment: deployment["deployed_at"],
                        default=None,
                    )
                    drift_status, actual_content_hash = (None, None)
                    if last and stage in ("preprod", "prod"):
                        drift_status, actual_content_hash = await _probe_drift(
                            dashboard_ref, last["environment_id"], last["content_hash"], config_manager
                        )
                    environments.append(
                        EnvironmentDeploymentStatus(
                            stage=stage,
                            commit_hash=last["commit_hash"] if last else None,
                            content_hash=last["content_hash"] if last else None,
                            deployed_at=last["deployed_at"] if last else None,
                            status="deployed" if last else "never_deployed",
                            is_behind=(last["content_hash"] != current_hash) if last and current_hash else None,
                            validation_status=last["validation_status"] if last else None,
                            validated_at=last["validated_at"] if last else None,
                            drift_status=drift_status,
                            actual_content_hash=actual_content_hash,
                            source_branch=last["source_branch"] if last else None,
                        )
                    )

                return DeploymentStatusResponse(
                    environments=environments,
                    current_content_hash=current_hash,
                )
            finally:
                db.close()

        except HTTPException:
            raise
        except Exception as e:
            _handle_unexpected_git_route_error("get_deployment_status", e)


# #endregion get_deployment_status


# #region validate_preprod_deployment [C:4] [TYPE Function] [SEMANTICS deployment, validation, preprod]
# @ingroup Api
# @BRIEF Mark the latest successfully deployed PREPROD content version as validated.
# @PRE The requested environment is PREPROD and has a successful deployment record.
# @POST The exact latest PREPROD deployment carries validation_status="validated".
# @SIDE_EFFECT Updates deployment_records validation metadata.
# @RATIONALE Validation is stored on the deployment record so publishing cannot be approved by a branch name alone.
# @REJECTED Storing a global PREPROD approval was rejected because a later deploy would incorrectly inherit it.
@router.post("/repositories/{dashboard_ref}/deployment-validation", response_model=DeploymentStatusResponse)
async def validate_preprod_deployment(
    dashboard_ref: str,
    payload: DeploymentValidationRequest,
    env_id: str | None = None,
    config_manager=Depends(get_config_manager),
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_user),
    _=Depends(has_permission("plugin:git", "EXECUTE")),
):
    with belief_scope("validate_preprod_deployment"):
        from . import _resolve_dashboard_id_from_ref
        from src.models.deployment import DeploymentRecord
        from src.models.git import DeploymentEnvironment

        try:
            dashboard_id = await _resolve_dashboard_id_from_ref(dashboard_ref, config_manager, env_id)
            repository = db.query(GitRepository).filter(GitRepository.dashboard_id == dashboard_id).first()
            if not repository:
                raise HTTPException(status_code=404, detail="Dashboard repository is not initialized")

            if payload.stage != "preprod":
                raise HTTPException(status_code=400, detail="Only the PREPROD environment can be validated")
            policy = config_manager.get_config().settings.git_release
            if policy.require_preprod_approval:
                _enforce_approval_policy(policy, current_user, payload.comment)
            environment = _resolve_stage_environment("preprod", db, config_manager)

            record = (
                db.query(DeploymentRecord)
                .filter(
                    DeploymentRecord.repository_id == repository.id,
                    DeploymentRecord.environment_id == environment.id,
                    DeploymentRecord.status == "success",
                )
                .order_by(DeploymentRecord.deployed_at.desc())
                .first()
            )
            if not record:
                raise HTTPException(status_code=409, detail="Deploy a dashboard version to PREPROD before validating it")

            record.validation_status = "validated"
            record.validated_at = datetime.now(UTC)
            record.validated_by = current_user.username
            db.commit()

            # Reuse the canonical status endpoint response after the atomic update.
            return await get_deployment_status(dashboard_ref, env_id, config_manager)
        except HTTPException:
            raise
        except Exception as e:
            db.rollback()
            _handle_unexpected_git_route_error("validate_preprod_deployment", e)


# #endregion validate_preprod_deployment


# #region deploy_dashboard [C:4] [TYPE Function] [SEMANTICS deployment, publish, validation]
# @ingroup Api
# @BRIEF Deploy dashboard from Git to a target environment.
# @POST PROD receives content only when the identical PREPROD deployment is validated.
# @RATIONALE The API enforces the same release gate as the UI, preventing direct calls from bypassing validation.
# @REJECTED UI-only publication gating was rejected because direct API calls could publish unvalidated content.
# @RELATION CALLS -> [GitPlugin]
@router.post("/repositories/{dashboard_ref}/deploy")
async def deploy_dashboard(
    dashboard_ref: str,
    deploy_data: DeployRequest,
    env_id: str | None = None,
    config_manager=Depends(get_config_manager),
    db: Session = Depends(get_db),
    _=Depends(has_permission("plugin:git", "EXECUTE")),
):
    with belief_scope("deploy_dashboard"):
        from . import _resolve_dashboard_id_from_ref
        from src.models.deployment import DeploymentRecord
        from src.models.git import DeploymentEnvironment
        from src.plugins.git_fingerprint import _compute_content_hash

        try:
            dashboard_id = await _resolve_dashboard_id_from_ref(dashboard_ref, config_manager, env_id)
            target = _resolve_stage_environment(deploy_data.stage, db, config_manager)
            if deploy_data.stage == "prod":
                policy = config_manager.get_config().settings.git_release
                repository = (
                    db.query(GitRepository)
                    .filter(GitRepository.dashboard_id == dashboard_id)
                    .first()
                )
                if not repository:
                    raise HTTPException(status_code=409, detail="Dashboard repository is not initialized")
                latest_preprod = (
                    db.query(DeploymentRecord)
                    .filter(
                        DeploymentRecord.repository_id == repository.id,
                        DeploymentRecord.environment_id == _resolve_stage_environment("preprod", db, config_manager).id,
                        DeploymentRecord.status == "success",
                    )
                    .order_by(DeploymentRecord.deployed_at.desc())
                    .first()
                )
                if latest_preprod and policy.block_publish_on_drift:
                    drift_status, _ = await _probe_drift(
                        dashboard_ref, latest_preprod.environment_id, latest_preprod.content_hash, config_manager
                    )
                    if drift_status != "in_sync":
                        raise HTTPException(
                            status_code=409,
                            detail="PREPROD version differs from the recorded release candidate; synchronize or redeploy before publishing",
                        )
                from src.services.git_service import GitService

                repo = await GitService().get_repo(dashboard_id)
                selected_commit = repo.commit(deploy_data.commit_hash).hexsha if deploy_data.commit_hash else None
                current_hash = _compute_content_hash(Path(repo.working_dir))
                if (
                    not latest_preprod
                    or (policy.require_prod_approval and latest_preprod.validation_status != "validated")
                    or (
                        policy.approval_expires_hours > 0
                        and latest_preprod.validated_at
                        and (datetime.now(UTC) - latest_preprod.validated_at.replace(tzinfo=UTC)).total_seconds() > policy.approval_expires_hours * 3600
                    )
                    or (
                        latest_preprod.commit_hash != selected_commit
                        if selected_commit
                        else latest_preprod.content_hash != current_hash
                    )
                ):
                    raise HTTPException(
                        status_code=409,
                        detail="Deploy the current dashboard content to PREPROD and validate it before publishing to PROD",
                    )
                source_branch = (latest_preprod.resources_changed or {}).get("source_branch")
            else:
                source_branch = deploy_data.source_branch
            from src.plugins.git_plugin import GitPlugin

            plugin = GitPlugin()
            result = await plugin.execute(
                {
                    "operation": "deploy",
                    "dashboard_id": dashboard_id,
                    "environment_id": target.id,
                    "commit_hash": deploy_data.commit_hash,
                    "source_branch": source_branch,
                }
            )
            if deploy_data.stage == "preprod":
                # PREPROD is a single shared slot: retain older candidates for audit,
                # but make only the newly deployed record eligible for approval/publish.
                db.expire_all()
                repository = db.query(GitRepository).filter(GitRepository.dashboard_id == dashboard_id).first()
                current_candidate = (
                    db.query(DeploymentRecord)
                    .filter(
                        DeploymentRecord.repository_id == repository.id,
                        DeploymentRecord.environment_id == target.id,
                        DeploymentRecord.status == "success",
                    )
                    .order_by(DeploymentRecord.deployed_at.desc(), DeploymentRecord.id.desc())
                    .first()
                ) if repository else None
                if current_candidate:
                    (
                        db.query(DeploymentRecord)
                        .filter(
                            DeploymentRecord.repository_id == repository.id,
                            DeploymentRecord.environment_id == target.id,
                            DeploymentRecord.status == "success",
                            DeploymentRecord.id != current_candidate.id,
                        )
                        .update({"status": "superseded", "validation_status": "superseded"}, synchronize_session=False)
                    )
                    db.commit()
            return result
        except HTTPException:
            raise
        except Exception as e:
            _handle_unexpected_git_route_error("deploy_dashboard", e)


# #endregion deploy_dashboard
# #endregion GitRepoLifecycleRoutes