agent-platform/services/workflow-service/app/domain/repositories.py

from datetime import datetime

from sqlalchemy import func, select
from sqlalchemy.orm import Session

from app.db.models import AppDefinition, AppVersion, WorkflowDefinitionModel, WorkflowVersion


class AppDefinitionRepository:
    def __init__(self, db: Session) -> None:
        self.db = db

    def create(
        self,
        *,
        tenant_id: str,
        code: str,
        name: str,
        description: str | None,
        owner_user_id: str | None,
        settings_json: dict | None,
    ) -> AppDefinition:
        entity = AppDefinition(
            tenant_id=tenant_id,
            code=code,
            name=name,
            description=description,
            owner_user_id=owner_user_id,
            settings_json=settings_json,
        )
        self.db.add(entity)
        self.db.commit()
        self.db.refresh(entity)
        return entity

    def list_by_tenant(self, tenant_id: str) -> list[AppDefinition]:
        stmt = select(AppDefinition).where(AppDefinition.tenant_id == tenant_id)
        return list(self.db.scalars(stmt))


class WorkflowDefinitionRepository:
    def __init__(self, db: Session) -> None:
        self.db = db

    def create(
        self,
        *,
        tenant_id: str,
        app_id: str,
        code: str,
        name: str,
        workflow_type: str,
    ) -> WorkflowDefinitionModel:
        entity = WorkflowDefinitionModel(
            tenant_id=tenant_id,
            app_id=app_id,
            code=code,
            name=name,
            workflow_type=workflow_type,
        )
        self.db.add(entity)
        self.db.commit()
        self.db.refresh(entity)
        return entity

    def list_by_scope(self, *, tenant_id: str, app_id: str | None = None) -> list[WorkflowDefinitionModel]:
        stmt = select(WorkflowDefinitionModel).where(WorkflowDefinitionModel.tenant_id == tenant_id)
        if app_id:
            stmt = stmt.where(WorkflowDefinitionModel.app_id == app_id)
        return list(self.db.scalars(stmt))


class AppVersionRepository:
    def __init__(self, db: Session) -> None:
        self.db = db

    def create(
        self,
        *,
        tenant_id: str,
        app_id: str,
        workflow_version_id: str,
        status: str,
        published_by: str | None,
        changelog: str | None,
    ) -> AppVersion:
        version_no = self._next_version_no(app_id)
        published_time = datetime.utcnow() if status == "published" else None
        entity = AppVersion(
            tenant_id=tenant_id,
            app_id=app_id,
            version_no=version_no,
            workflow_version_id=workflow_version_id,
            status=status,
            published_by=published_by,
            published_time=published_time,
            changelog=changelog,
        )
        self.db.add(entity)
        self.db.commit()
        self.db.refresh(entity)
        return entity

    def list_by_app(self, *, tenant_id: str, app_id: str) -> list[AppVersion]:
        stmt = (
            select(AppVersion)
            .where(AppVersion.tenant_id == tenant_id)
            .where(AppVersion.app_id == app_id)
            .order_by(AppVersion.version_no.desc())
        )
        return list(self.db.scalars(stmt))

    def _next_version_no(self, app_id: str) -> int:
        stmt = select(func.max(AppVersion.version_no)).where(AppVersion.app_id == app_id)
        current_max = self.db.scalar(stmt)
        return (current_max or 0) + 1


class WorkflowVersionRepository:
    def __init__(self, db: Session) -> None:
        self.db = db

    def create(
        self,
        *,
        tenant_id: str,
        workflow_id: str,
        dsl_json: dict | None,
        compiled_plan_json: dict | None,
        schema_version: str | None,
        checksum: str | None,
        status: str,
    ) -> WorkflowVersion:
        version_no = self._next_version_no(workflow_id)
        entity = WorkflowVersion(
            tenant_id=tenant_id,
            workflow_id=workflow_id,
            version_no=version_no,
            dsl_json=dsl_json,
            compiled_plan_json=compiled_plan_json,
            schema_version=schema_version,
            checksum=checksum,
            status=status,
        )
        self.db.add(entity)
        self.db.commit()
        self.db.refresh(entity)
        return entity

    def list_by_workflow(self, *, tenant_id: str, workflow_id: str) -> list[WorkflowVersion]:
        stmt = (
            select(WorkflowVersion)
            .where(WorkflowVersion.tenant_id == tenant_id)
            .where(WorkflowVersion.workflow_id == workflow_id)
            .order_by(WorkflowVersion.version_no.desc())
        )
        return list(self.db.scalars(stmt))

    def get_by_id(self, *, tenant_id: str, workflow_version_id: str) -> WorkflowVersion | None:
        stmt = (
            select(WorkflowVersion)
            .where(WorkflowVersion.tenant_id == tenant_id)
            .where(WorkflowVersion.id == workflow_version_id)
        )
        return self.db.scalar(stmt)

    def _next_version_no(self, workflow_id: str) -> int:
        stmt = select(func.max(WorkflowVersion.version_no)).where(
            WorkflowVersion.workflow_id == workflow_id
        )
        current_max = self.db.scalar(stmt)
        return (current_max or 0) + 1