from __future__ import annotations import base64 import binascii import json import time from datetime import datetime, timezone from typing import Any from uuid import UUID from app.core.logger import get_logger from app.generation.models import GenerationJob from app.projects.errors import ( ProjectInvalidCursorError, ProjectInvalidNameError, ProjectInvalidStatusError, ProjectThumbnailInvalidError, ) from app.projects.models import Project, ProjectGenerationJob from app.projects.repositories.project_repository import ProjectRepository from app.projects.schemas import ( ProjectAssetListResponse, ProjectAssetResponse, ProjectCreate, ProjectGenerationJobListResponse, ProjectGenerationJobResponse, ProjectListResponse, ProjectResponse, ProjectStatus, ProjectUpdate, normalize_project_name, ) from app.brand.repositories.brand_repository import BrandKitRepository from app.security.assets import CanonicalAssetNotFoundError, CanonicalAssetService from app.security.audit import AuditService from app.security.models import CanonicalMediaAsset logger = get_logger(__name__) class ProjectService: """Coordinates validation, tenant ownership, persistence, and audit.""" def __init__( self, repository: ProjectRepository, assets: CanonicalAssetService, audit: AuditService, brand_kits: BrandKitRepository, ) -> None: self.repository = repository self.assets = assets self.audit = audit self.brand_kits = brand_kits async def get_active_brand_kit(self, *, workspace_id: str, project_id: str, user_id: str) -> Any | None: project = await self.repository.get(workspace_id, project_id, user_id=user_id) if not project.brand_kit_id: return None return await self.brand_kits.get_latest_version(workspace_id, project.brand_kit_id, user_id=user_id) async def create( self, *, workspace_id: str, user_id: str, api_key_id: str, request_id: str, payload: ProjectCreate, ) -> ProjectResponse: started = time.monotonic() try: name = normalize_project_name(payload.name) except ValueError as exc: raise ProjectInvalidNameError(str(exc)) from exc await self._validate_thumbnail(workspace_id, user_id, payload.thumbnail_asset_id) project = await self.repository.create( Project( workspace_id=workspace_id, created_by=user_id, name=name, description=payload.description, status=ProjectStatus.ACTIVE.value, thumbnail_asset_id=payload.thumbnail_asset_id, metadata_json=dict(payload.metadata), ), user_id=user_id, ) await self._audit( project, user_id=user_id, api_key_id=api_key_id, request_id=request_id, event_type="project.created", metadata={"has_thumbnail": project.thumbnail_asset_id is not None}, ) self._log("create", project, started, "success") return self._response(project) async def list( self, *, workspace_id: str, user_id: str, status: ProjectStatus | None, search: str | None, limit: int, cursor: str | None, ) -> ProjectListResponse: cursor_updated_at, cursor_id = self._decode_cursor(cursor) projects, has_more = await self.repository.list( workspace_id, user_id=user_id, status=status, search=search.strip() if search else None, limit=limit, cursor_updated_at=cursor_updated_at, cursor_id=cursor_id, ) next_cursor = self._encode_cursor(projects[-1]) if has_more and projects else None # NOTE: For listing, we might want to optimize this to avoid N+1 queries. # Keeping it simple for now as requested. responses = [] for project in projects: active_version = None if project.brand_kit_id: active_version = await self.brand_kits.get_latest_version(workspace_id, project.brand_kit_id, user_id=user_id) responses.append(self._response(project, active_version.id if active_version else None)) return ProjectListResponse( items=responses, next_cursor=next_cursor, limit=limit, ) async def get(self, *, workspace_id: str, user_id: str, project_id: str) -> ProjectResponse: project = await self.repository.get(workspace_id, project_id, user_id=user_id) active_version = None if project.brand_kit_id: active_version = await self.brand_kits.get_latest_version(workspace_id, project.brand_kit_id, user_id=user_id) return self._response(project, active_version.id if active_version else None) async def update( self, *, workspace_id: str, user_id: str, api_key_id: str, request_id: str, project_id: str, payload: ProjectUpdate, ) -> ProjectResponse: started = time.monotonic() fields = payload.model_dump(exclude_unset=True) if "name" in fields: try: fields["name"] = normalize_project_name(str(fields["name"])) except ValueError as exc: raise ProjectInvalidNameError(str(exc)) from exc if "thumbnail_asset_id" in fields: await self._validate_thumbnail(workspace_id, user_id, fields["thumbnail_asset_id"]) if "metadata" in fields: fields["metadata_json"] = fields.pop("metadata") if "status" in fields: status = fields["status"] try: fields["status"] = ProjectStatus(status).value except ValueError as exc: raise ProjectInvalidStatusError("Project status is invalid.") from exc project = await self.repository.update( workspace_id, project_id, user_id=user_id, fields=fields, ) archived = fields.get("status") == ProjectStatus.ARCHIVED.value await self._audit( project, user_id=user_id, api_key_id=api_key_id, request_id=request_id, event_type="project.archived" if archived else "project.updated", metadata={"fields": sorted(payload.model_fields_set)}, ) self._log("update", project, started, "success") return self._response(project) async def delete( self, *, workspace_id: str, user_id: str, api_key_id: str, request_id: str, project_id: str, ) -> None: started = time.monotonic() project = await self.repository.archive(workspace_id, project_id, user_id=user_id) await self._audit( project, user_id=user_id, api_key_id=api_key_id, request_id=request_id, event_type="project.deleted", metadata={"disposition": "archived"}, ) self._log("delete", project, started, "success") async def list_assets( self, *, workspace_id: str, user_id: str, project_id: str ) -> ProjectAssetListResponse: assets = await self.repository.list_assets(workspace_id, project_id, user_id=user_id) return ProjectAssetListResponse(items=[self._asset_response(asset) for asset in assets]) async def attach_asset( self, *, workspace_id: str, user_id: str, api_key_id: str, request_id: str, project_id: str, asset_id: str, ) -> ProjectAssetResponse: asset, attached = await self.repository.attach_asset( workspace_id, project_id, asset_id, user_id=user_id ) if attached: await self._audit_resource( workspace_id=workspace_id, user_id=user_id, api_key_id=api_key_id, request_id=request_id, project_id=project_id, event_type="project.asset_attached", resource_id=asset.id, ) return self._asset_response(asset) async def detach_asset( self, *, workspace_id: str, user_id: str, api_key_id: str, request_id: str, project_id: str, asset_id: str, ) -> None: asset = await self.repository.detach_asset( workspace_id, project_id, asset_id, user_id=user_id ) await self._audit_resource( workspace_id=workspace_id, user_id=user_id, api_key_id=api_key_id, request_id=request_id, project_id=project_id, event_type="project.asset_detached", resource_id=asset.id, ) async def list_generation_jobs( self, *, workspace_id: str, user_id: str, project_id: str ) -> ProjectGenerationJobListResponse: jobs = await self.repository.list_generation_jobs(workspace_id, project_id, user_id=user_id) return ProjectGenerationJobListResponse( items=[self._job_response(association, job) for association, job in jobs] ) async def attach_generation_job( self, *, workspace_id: str, user_id: str, api_key_id: str, request_id: str, project_id: str, generation_job_id: str, ) -> ProjectGenerationJobResponse: association, job, attached = await self.repository.attach_generation_job( workspace_id, project_id, generation_job_id, user_id=user_id ) if attached: await self._audit_resource( workspace_id=workspace_id, user_id=user_id, api_key_id=api_key_id, request_id=request_id, project_id=project_id, event_type="project.job_attached", resource_id=job.id, ) return self._job_response(association, job) async def detach_generation_job( self, *, workspace_id: str, user_id: str, api_key_id: str, request_id: str, project_id: str, generation_job_id: str, ) -> None: _, job = await self.repository.detach_generation_job( workspace_id, project_id, generation_job_id, user_id=user_id ) await self._audit_resource( workspace_id=workspace_id, user_id=user_id, api_key_id=api_key_id, request_id=request_id, project_id=project_id, event_type="project.job_detached", resource_id=job.id, ) async def _validate_thumbnail(self, workspace_id: str, user_id: str, asset_id: Any) -> None: if asset_id is None: return if not isinstance(asset_id, str): raise ProjectThumbnailInvalidError("Project thumbnail reference is invalid.") try: await self.assets.get_owned_by_id( workspace_id=workspace_id, user_id=user_id, asset_id=asset_id ) except CanonicalAssetNotFoundError as exc: raise ProjectThumbnailInvalidError( "Project thumbnail is not an owned canonical asset." ) from exc async def _audit( self, project: Project, *, user_id: str, api_key_id: str, request_id: str, event_type: str, metadata: dict[str, object], ) -> None: await self.audit.record_event( workspace_id=project.workspace_id, user_id=user_id, api_key_id=api_key_id, request_id=request_id, event_type=event_type, entity_type="project", entity_id=project.id, metadata=metadata, ) async def _audit_resource( self, *, workspace_id: str, user_id: str, api_key_id: str, request_id: str, project_id: str, event_type: str, resource_id: str, ) -> None: await self.audit.record_event( workspace_id=workspace_id, user_id=user_id, api_key_id=api_key_id, request_id=request_id, event_type=event_type, entity_type="project", entity_id=project_id, metadata={"resource_id": resource_id}, ) @staticmethod def _response(project: Project, active_version_id: str | None = None) -> ProjectResponse: return ProjectResponse( id=project.id, workspace_id=project.workspace_id, name=project.name, description=project.description, status=ProjectStatus(project.status), thumbnail_asset_id=project.thumbnail_asset_id, brand_kit_id=project.brand_kit_id, brand_kit_version_id=active_version_id, metadata=dict(project.metadata_json or {}), created_by=project.created_by, created_at=project.created_at, updated_at=project.updated_at, archived_at=project.archived_at, ) @staticmethod def _asset_response(asset: CanonicalMediaAsset) -> ProjectAssetResponse: return ProjectAssetResponse( id=asset.id, project_id=str(asset.project_id), request_id=asset.request_id, filename=asset.filename, mime_type=asset.mime_type, file_size=asset.file_size, metadata=dict(asset.metadata_json or {}), created_by_user_id=asset.created_by_user_id, created_at=asset.created_at, updated_at=asset.updated_at, ) @staticmethod def _job_response( association: ProjectGenerationJob, job: GenerationJob ) -> ProjectGenerationJobResponse: return ProjectGenerationJobResponse( id=job.id, project_id=association.project_id, generation_request_id=job.generation_request_id, provider=job.provider, status=job.status, output_asset_id=job.output_asset_id, error_code=job.error_code, created_at=job.created_at, started_at=job.started_at, completed_at=job.completed_at, updated_at=job.updated_at, attached_at=association.created_at, ) @staticmethod def _encode_cursor(project: Project) -> str: payload = json.dumps( {"updated_at": project.updated_at.isoformat(), "id": project.id}, separators=(",", ":"), ).encode("utf-8") return base64.urlsafe_b64encode(payload).decode("ascii").rstrip("=") @staticmethod def _decode_cursor(cursor: str | None) -> tuple[datetime | None, str | None]: if cursor is None: return None, None try: padding = "=" * (-len(cursor) % 4) payload = json.loads(base64.urlsafe_b64decode(cursor + padding)) updated_at = datetime.fromisoformat(payload["updated_at"]) if updated_at.tzinfo is None: updated_at = updated_at.replace(tzinfo=timezone.utc) project_id = payload["id"] if not isinstance(project_id, str): raise ValueError UUID(project_id) return updated_at, project_id except (KeyError, TypeError, ValueError, binascii.Error, json.JSONDecodeError) as exc: raise ProjectInvalidCursorError("Project pagination cursor is invalid.") from exc @staticmethod def _log( operation: str, project: Project, started: float, result: str, ) -> None: logger.info( "project operation completed", extra={ "operation": operation, "project_id": project.id, "workspace_id": project.workspace_id, "duration_ms": max(0, round((time.monotonic() - started) * 1000)), "result": result, }, )