"""project_storage domain — business logic for bucket management and usage tracking.""" from __future__ import annotations from typing import Optional from uuid import UUID import structlog from src.core.exceptions import ConflictError, NotFoundError from src.domain.project_storage.models import ProjectStorageBucket, ProjectStorageUsage from src.domain.project_storage.repository import ( ProjectStorageBucketRepository, ProjectStorageUsageRepository, ) from src.domain.project_storage.schemas import ( BucketResponse, CreateBucketRequest, StorageUsageAckResponse, StorageUsageReportRequest, StorageUsageResponse, UpdateBucketRequest, ) from src.domain.projects.repository import ProjectRepository from src.integrations.cloud_control_client import CloudControlClient logger = structlog.get_logger() # Max buckets per project — soft limit before plan entitlements are wired up _DEFAULT_MAX_BUCKETS = 10 class ProjectStorageService: def __init__( self, *, bucket_repo: ProjectStorageBucketRepository, usage_repo: ProjectStorageUsageRepository, cloud_client: CloudControlClient, ) -> None: self._buckets = bucket_repo self._usage = usage_repo self._cloud = cloud_client # ------------------------------------------------------------------ # Bucket management # ------------------------------------------------------------------ async def list_buckets(self, project_id: UUID) -> list[BucketResponse]: rows = await self._buckets.list_for_project(project_id) return [_bucket_response(r) for r in rows] async def create_bucket( self, project_id: UUID, body: CreateBucketRequest ) -> BucketResponse: existing = await self._buckets.get_by_name(project_id, body.name) if existing: raise ConflictError(f"Bucket {body.name!r} already exists for this project") count = await self._buckets.count_active(project_id) if count >= _DEFAULT_MAX_BUCKETS: raise ConflictError( f"Project has reached the maximum of {_DEFAULT_MAX_BUCKETS} buckets" ) row = await self._buckets.create( project_id=project_id, name=body.name, access_policy=body.access_policy, rls_enabled=body.rls_enabled, cdn_enabled=body.cdn_enabled, cdn_custom_domain=body.cdn_custom_domain, allowed_mime_types_json=body.allowed_mime_types, max_file_size_bytes=body.max_file_size_bytes, ) logger.info("storage_bucket_created", project_id=str(project_id), bucket=body.name) await self._push_buckets(project_id) return _bucket_response(row) async def update_bucket( self, project_id: UUID, bucket_name: str, body: UpdateBucketRequest ) -> BucketResponse: row = await self._buckets.get_by_name(project_id, bucket_name) if not row: raise NotFoundError(f"Bucket {bucket_name!r} not found") fields: dict = {} if body.access_policy is not None: fields["access_policy"] = body.access_policy if body.rls_enabled is not None: fields["rls_enabled"] = body.rls_enabled if body.cdn_enabled is not None: fields["cdn_enabled"] = body.cdn_enabled if body.cdn_custom_domain is not None: fields["cdn_custom_domain"] = body.cdn_custom_domain or None if body.allowed_mime_types is not None: fields["allowed_mime_types_json"] = body.allowed_mime_types if body.max_file_size_bytes is not None: fields["max_file_size_bytes"] = body.max_file_size_bytes if body.is_active is not None: fields["is_active"] = body.is_active row = await self._buckets.update(row, **fields) await self._push_buckets(project_id) return _bucket_response(row) async def sync_buckets_to_cloud(self, project_id: UUID) -> None: """Re-push active bucket definitions to Cloud D1 (idempotent).""" await self._push_buckets(project_id) async def delete_bucket(self, project_id: UUID, bucket_name: str) -> None: """Soft-delete by marking inactive. Actual R2 cleanup is deferred to ops.""" row = await self._buckets.get_by_name(project_id, bucket_name) if not row: raise NotFoundError(f"Bucket {bucket_name!r} not found") await self._buckets.update(row, is_active=False) await self._push_buckets(project_id) logger.info("storage_bucket_deleted", project_id=str(project_id), bucket=bucket_name) async def get_usage(self, project_id: UUID) -> StorageUsageResponse: row = await self._usage.get_for_project(project_id) if not row: return StorageUsageResponse( project_id=project_id, total_bytes=0, total_files=0, last_updated_at=None ) return StorageUsageResponse( project_id=row.project_id, total_bytes=row.total_bytes, total_files=row.total_files, last_updated_at=row.last_updated_at, ) # ------------------------------------------------------------------ # Usage reporting (called from Cloud → Core usage endpoint) # ------------------------------------------------------------------ async def apply_usage_report( self, body: StorageUsageReportRequest ) -> StorageUsageAckResponse: """Apply a storage delta from Cloud and update the running counter.""" try: project_id = UUID(body.project_id) except ValueError: raise NotFoundError("Invalid project_id") row = await self._usage.apply_delta( project_id=project_id, delta_bytes=body.delta_bytes, delta_files=body.delta_files, ) logger.info( "storage_usage_updated", project_id=body.project_id, event=body.event, delta_bytes=body.delta_bytes, total_bytes=row.total_bytes, ) return StorageUsageAckResponse( ok=True, total_bytes=row.total_bytes, total_files=row.total_files, ) # ------------------------------------------------------------------ # Control plane push # ------------------------------------------------------------------ async def _push_buckets(self, project_id: UUID) -> None: project = await ProjectRepository(self._buckets._db).get_by_id(project_id) if not project: logger.warning("storage_bucket_push_skip", reason="project_not_found", project_id=str(project_id)) return rows = await self._buckets.list_for_project(project_id) buckets = [ { "id": str(r.id), "name": r.name, "access_policy": r.access_policy, "rls_enabled": r.rls_enabled, "cdn_enabled": r.cdn_enabled, "cdn_url_base": r.cdn_custom_domain, "allowed_mime_types": r.allowed_mime_types_json, "max_file_size_bytes": r.max_file_size_bytes, } for r in rows if r.is_active ] await self._cloud.push_storage_buckets( project_id=str(project_id), workspace_id=str(project.workspace_id), buckets=buckets, ) # --------------------------------------------------------------------------- # Response builders # --------------------------------------------------------------------------- def _bucket_response(row: ProjectStorageBucket) -> BucketResponse: return BucketResponse( id=row.id, project_id=row.project_id, name=row.name, access_policy=row.access_policy, rls_enabled=row.rls_enabled, cdn_enabled=row.cdn_enabled, cdn_custom_domain=row.cdn_custom_domain, allowed_mime_types=row.allowed_mime_types_json, max_file_size_bytes=row.max_file_size_bytes, is_active=row.is_active, created_at=row.created_at, updated_at=row.updated_at, )