smart-org/apps/api/app/infrastructure/repositories/projects.py

"""Repositories for Projects, their Git repositories, credentials and preview
domains."""

from __future__ import annotations

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.domain.projects.models import (
    Project,
    ProjectGitCredential,
    ProjectGitRepo,
)
from app.infrastructure.db.models import (
    PreviewDomainRow,
    PreviewPortRow,
    ProjectGitCredentialRow,
    ProjectRepositoryRow,
    ProjectRow,
)


def _project_to_domain(row: ProjectRow) -> Project:
    return Project(
        id=row.id, organization_id=row.organization_id, name=row.name, slug=row.slug,
        description=row.description, created_by_user_id=row.created_by_user_id,
        default_branch=row.default_branch, created_at=row.created_at,
    )


def _repo_to_domain(row: ProjectRepositoryRow) -> ProjectGitRepo:
    return ProjectGitRepo(
        id=row.id, project_id=row.project_id, provider=row.provider, owner=row.owner,
        repo_name=row.repo_name, clone_url=row.clone_url, web_url=row.web_url,
        default_branch=row.default_branch, created_at=row.created_at,
    )


class ProjectGitRepoRepository:
    """Persistence for the Git repository attached to a project."""

    def __init__(self, session: AsyncSession) -> None:
        self.session = session

    async def get_for_project(self, project_id: str) -> ProjectGitRepo | None:
        stmt = select(ProjectRepositoryRow).where(
            ProjectRepositoryRow.project_id == project_id
        )
        row = (await self.session.execute(stmt)).scalar_one_or_none()
        return _repo_to_domain(row) if row else None

    async def upsert(self, repo: ProjectGitRepo) -> ProjectGitRepo:
        existing = await self.get_for_project(repo.project_id)
        if existing:
            row = await self.session.get(ProjectRepositoryRow, existing.id)
            assert row is not None
            row.provider = repo.provider
            row.owner = repo.owner
            row.repo_name = repo.repo_name
            row.clone_url = repo.clone_url
            row.web_url = repo.web_url
            row.default_branch = repo.default_branch
            await self.session.flush()
            return _repo_to_domain(row)

        row = ProjectRepositoryRow(
            id=repo.id, project_id=repo.project_id, provider=repo.provider,
            owner=repo.owner, repo_name=repo.repo_name, clone_url=repo.clone_url,
            web_url=repo.web_url, default_branch=repo.default_branch,
            created_at=repo.created_at,
        )
        self.session.add(row)
        await self.session.flush()
        return _repo_to_domain(row)


class ProjectGitCredentialRepository:
    def __init__(self, session: AsyncSession) -> None:
        self.session = session

    async def get_for_project(self, project_id: str) -> ProjectGitCredential | None:
        stmt = select(ProjectGitCredentialRow).where(
            ProjectGitCredentialRow.project_id == project_id
        )
        row = (await self.session.execute(stmt)).scalar_one_or_none()
        if row is None:
            return None
        return ProjectGitCredential(
            id=row.id, project_id=row.project_id, username=row.username,
            encrypted_password=row.encrypted_password, created_at=row.created_at,
        )

    async def upsert(self, credential: ProjectGitCredential) -> ProjectGitCredential:
        existing = await self.get_for_project(credential.project_id)
        if existing:
            row = await self.session.get(ProjectGitCredentialRow, existing.id)
            assert row is not None
            row.username = credential.username
            row.encrypted_password = credential.encrypted_password
            await self.session.flush()
            return ProjectGitCredential(
                id=row.id, project_id=row.project_id, username=row.username,
                encrypted_password=row.encrypted_password, created_at=row.created_at,
            )

        row = ProjectGitCredentialRow(
            id=credential.id, project_id=credential.project_id,
            username=credential.username,
            encrypted_password=credential.encrypted_password,
            created_at=credential.created_at,
        )
        self.session.add(row)
        await self.session.flush()
        return credential


class ProjectRepository:
    def __init__(self, session: AsyncSession) -> None:
        self.session = session

    async def get(self, project_id: str) -> Project | None:
        row = await self.session.get(ProjectRow, project_id)
        return _project_to_domain(row) if row else None

    async def list_for_org(
        self, organization_id: str, *, limit: int = 100, offset: int = 0
    ) -> list[Project]:
        stmt = (
            select(ProjectRow)
            .where(ProjectRow.organization_id == organization_id)
            .order_by(ProjectRow.created_at.desc())
            .limit(min(limit, 500))
            .offset(offset)
        )
        return [_project_to_domain(r) for r in (await self.session.execute(stmt)).scalars()]

    async def list_slugs(self, organization_id: str) -> set[str]:
        stmt = select(ProjectRow.slug).where(ProjectRow.organization_id == organization_id)
        return set((await self.session.execute(stmt)).scalars().all())

    async def create(self, project: Project) -> Project:
        row = ProjectRow(
            id=project.id, organization_id=project.organization_id, name=project.name,
            slug=project.slug, description=project.description,
            created_by_user_id=project.created_by_user_id,
            default_branch=project.default_branch, created_at=project.created_at,
        )
        self.session.add(row)
        await self.session.flush()
        return _project_to_domain(row)

    async def delete(self, project_id: str) -> None:
        row = await self.session.get(ProjectRow, project_id)
        if row:
            await self.session.delete(row)
            await self.session.flush()


class PreviewDomainRepository:
    def __init__(self, session: AsyncSession) -> None:
        self.session = session

    # -- ports ---------------------------------------------------------------

    async def get_port(self, project_id: str, service_name: str) -> int | None:
        stmt = select(PreviewPortRow.port).where(
            PreviewPortRow.project_id == project_id,
            PreviewPortRow.service_name == service_name,
        )
        return (await self.session.execute(stmt)).scalar_one_or_none()

    async def list_used_ports(self, *, start: int, end: int) -> set[int]:
        stmt = select(PreviewPortRow.port).where(
            PreviewPortRow.port.between(start, end)
        )
        return set((await self.session.execute(stmt)).scalars().all())

    async def add_port(
        self, *, organization_id: str, project_id: str, service_name: str, port: int
    ) -> None:
        self.session.add(
            PreviewPortRow(
                organization_id=organization_id, project_id=project_id,
                service_name=service_name, port=port, status="active",
            )
        )
        await self.session.flush()

    # -- domains -------------------------------------------------------------

    async def get_by_host(self, host: str) -> dict | None:
        stmt = select(PreviewDomainRow).where(PreviewDomainRow.host == host)
        row = (await self.session.execute(stmt)).scalar_one_or_none()
        return self._row_to_dict(row) if row else None

    async def get(self, domain_id: str) -> dict | None:
        row = await self.session.get(PreviewDomainRow, domain_id)
        return self._row_to_dict(row) if row else None

    async def list_for_project(self, project_id: str) -> list[dict]:
        stmt = (
            select(PreviewDomainRow)
            .where(PreviewDomainRow.project_id == project_id)
            .order_by(PreviewDomainRow.created_at)
        )
        return [self._row_to_dict(r) for r in (await self.session.execute(stmt)).scalars()]

    async def count_for_project(self, project_id: str) -> int:
        return len(await self.list_for_project(project_id))

    async def create(
        self,
        *,
        organization_id: str,
        project_id: str,
        host: str,
        port: int,
        created_by_user_id: str | None,
    ) -> dict:
        row = PreviewDomainRow(
            organization_id=organization_id, project_id=project_id, host=host,
            port=port, status="active", protected=False,
            created_by_user_id=created_by_user_id,
        )
        self.session.add(row)
        await self.session.flush()
        return self._row_to_dict(row)

    async def set_status(self, domain_id: str, status: str) -> None:
        row = await self.session.get(PreviewDomainRow, domain_id)
        if row:
            row.status = status
            await self.session.flush()

    async def delete(self, domain_id: str) -> None:
        row = await self.session.get(PreviewDomainRow, domain_id)
        if row:
            await self.session.delete(row)
            await self.session.flush()

    @staticmethod
    def _row_to_dict(row: PreviewDomainRow) -> dict:
        return {
            "id": row.id,
            "organization_id": row.organization_id,
            "project_id": row.project_id,
            "host": row.host,
            "port": row.port,
            "status": row.status,
            "protected": row.protected,
            "created_by_user_id": row.created_by_user_id,
            "created_at": row.created_at,
        }