"""Enqueue Dropbox imports via P2HS Cloud and list resulting storage assets.""" from __future__ import annotations import hashlib import json import uuid from uuid import UUID from sqlalchemy.ext.asyncio import AsyncSession from src.domain.cloud.schemas import ( AuthorizeRequest, AuthorizeRequestMeta, AuthorizeResponseAllowed, AuthorizeResponseDenied, ) from src.domain.cloud.service import CloudInternalService, _expected_client_id from src.domain.dropbox_imports.schemas import ( DROPBOX_IMPORT_BUCKET, ArchiveExtractJobResponse, AssetAccessUrlResponse, DeleteStorageAssetResponse, DropboxBucketSettingsResponse, DropboxImportJobResponse, StartArchiveExtractRequest, StartDropboxImportRequest, StorageAssetResponse, StorageAssetsListResponse, UpdateDropboxBucketSettingsRequest, ) from src.domain.project_storage.repository import ProjectStorageBucketRepository from src.domain.project_storage.schemas import CreateBucketRequest, UpdateBucketRequest from src.domain.project_storage.service import ProjectStorageService from src.domain.projects.models import Project from src.domain.projects.repository import ProjectRepository from src.integrations.cloud_control_client import get_cloud_client_for_workspace def _is_archive_file(*, filename: str, content_type: str | None) -> bool: name = filename.lower() ct = (content_type or "").lower() return name.endswith(".zip") or "zip" in ct def _dashboard_poll_url(project_id: UUID, job_id: str) -> str: return f"/api/projects/{project_id}/jobs/{job_id}" def _bucket_settings_from_row(bucket_name: str, row) -> DropboxBucketSettingsResponse: if row is None: return DropboxBucketSettingsResponse( bucket_name=bucket_name, access_policy="private", cdn_enabled=False, cdn_base_url=None, public_urls_enabled=False, ) access_policy = getattr(row, "access_policy", "private") cdn_enabled = bool(getattr(row, "cdn_enabled", False)) cdn_base = getattr(row, "cdn_custom_domain", None) public = access_policy == "public" and cdn_enabled and bool(cdn_base) return DropboxBucketSettingsResponse( bucket_name=bucket_name, access_policy=access_policy, cdn_enabled=cdn_enabled, cdn_base_url=cdn_base, public_urls_enabled=public, ) class DropboxImportService: def __init__( self, *, db: AsyncSession, project_repo: ProjectRepository, cloud_service: CloudInternalService, storage_service: ProjectStorageService, bucket_repo: ProjectStorageBucketRepository, ) -> None: self._db = db self._project_repo = project_repo self._cloud_service = cloud_service self._storage_service = storage_service self._bucket_repo = bucket_repo async def _ensure_bucket(self, project_id: UUID, bucket_name: str) -> None: """Ensure bucket exists in Core PG only; Cloud D1 bucket is provisioned on job enqueue.""" existing = await self._bucket_repo.get_by_name(project_id, bucket_name) if existing: if not existing.is_active: await self._storage_service.update_bucket( project_id, bucket_name, UpdateBucketRequest(is_active=True) ) return try: await self._storage_service.create_bucket( project_id, CreateBucketRequest(name=bucket_name, access_policy="private"), ) except Exception: # Bucket may exist in PG but Cloud sync failed; job enqueue still provisions D1. if not await self._bucket_repo.get_by_name(project_id, bucket_name): raise async def start_import( self, project: Project, body: StartDropboxImportRequest, ) -> DropboxImportJobResponse: await self._ensure_bucket(project.id, body.bucket_name) url_str = body.dropbox_url payload_hash = hashlib.sha256(json.dumps({"url": url_str}, sort_keys=True).encode()).hexdigest() idem = body.idempotency_key or f"dropbox_{payload_hash[:40]}" auth = await self._cloud_service.authorize( AuthorizeRequest( workspace_id=str(project.workspace_id), project_id=str(project.id), client_id=_expected_client_id(project), handshake_id="core_dashboard", request_type="dropbox.import", idempotency_key=idem, request_meta=AuthorizeRequestMeta( origin="", payload_hash=f"sha256:{payload_hash}", estimated_units=1, provider_hint="dropbox", ), ) ) if isinstance(auth, AuthorizeResponseDenied): raise ValueError(auth.message or auth.reason_code) if not isinstance(auth, AuthorizeResponseAllowed): raise ValueError("authorization_failed") cloud_client = await get_cloud_client_for_workspace(project.workspace_id, self._db) result = await cloud_client.enqueue_control_job( project_id=str(project.id), workspace_id=str(project.workspace_id), client_id=_expected_client_id(project), request_type="dropbox.import", idempotency_key=idem, payload={ "dropbox_url": url_str, "bucket_name": body.bucket_name, "extract_archive": body.extract_archive, }, authorization_id=auth.authorization_id, grant_jwt=auth.grant.jwt, grant_expires_at=auth.grant.expires_at, policy_snapshot=auth.policy_snapshot.model_dump(mode="json"), ) if not result.get("ok"): raise RuntimeError(result.get("error") or "cloud_enqueue_failed") job_id = str(result.get("job_id") or "") return DropboxImportJobResponse( job_id=job_id, status=str(result.get("status") or "queued"), poll_url=_dashboard_poll_url(project.id, job_id) if job_id else None, ) async def start_archive_extract( self, project: Project, file_id: str, body: StartArchiveExtractRequest, ) -> ArchiveExtractJobResponse: """Enqueue billable zip extraction for an existing storage file.""" await self._ensure_bucket(project.id, body.bucket_name) cloud_client = await get_cloud_client_for_workspace(project.workspace_id, self._db) stored = await cloud_client.get_storage_file( project_id=str(project.id), bucket_name=body.bucket_name, file_id=file_id, ) if not stored: raise ValueError("file_not_found") if str(stored.get("status") or "") != "active": raise ValueError("file_not_active") filename = str(stored.get("filename") or "") content_type = stored.get("content_type") if not _is_archive_file(filename=filename, content_type=content_type): raise ValueError("file_not_zip_archive") idem = body.idempotency_key or f"extract_{file_id}_{uuid.uuid4().hex[:16]}" payload_hash = hashlib.sha256( json.dumps( {"file_id": file_id, "bucket": body.bucket_name}, sort_keys=True, ).encode(), ).hexdigest() auth = await self._cloud_service.authorize( AuthorizeRequest( workspace_id=str(project.workspace_id), project_id=str(project.id), client_id=_expected_client_id(project), handshake_id="core_dashboard", request_type="storage.extract", idempotency_key=idem, request_meta=AuthorizeRequestMeta( origin="", payload_hash=f"sha256:{payload_hash}", estimated_units=1, provider_hint="storage", ), ) ) if isinstance(auth, AuthorizeResponseDenied): raise ValueError(auth.message or auth.reason_code) if not isinstance(auth, AuthorizeResponseAllowed): raise ValueError("authorization_failed") result = await cloud_client.enqueue_control_job( project_id=str(project.id), workspace_id=str(project.workspace_id), client_id=_expected_client_id(project), request_type="storage.extract", idempotency_key=idem, payload={ "bucket_name": body.bucket_name, "source_file_id": file_id, "entry_offset": body.entry_offset, }, authorization_id=auth.authorization_id, grant_jwt=auth.grant.jwt, grant_expires_at=auth.grant.expires_at, policy_snapshot=auth.policy_snapshot.model_dump(mode="json"), ) if not result.get("ok"): raise RuntimeError(result.get("error") or "cloud_enqueue_failed") job_id = str(result.get("job_id") or "") return ArchiveExtractJobResponse( job_id=job_id, status=str(result.get("status") or "queued"), poll_url=_dashboard_poll_url(project.id, job_id) if job_id else None, source_file_id=file_id, ) async def list_assets( self, project: Project, *, bucket_name: str = DROPBOX_IMPORT_BUCKET, limit: int = 100, cursor: str | None = None, ) -> StorageAssetsListResponse: bucket_row = await self._bucket_repo.get_by_name(project.id, bucket_name) settings = _bucket_settings_from_row(bucket_name, bucket_row) cloud_client = await get_cloud_client_for_workspace(project.workspace_id, self._db) raw = await cloud_client.list_storage_files( project_id=str(project.id), bucket_name=bucket_name, limit=limit, cursor=cursor, ) files = [ StorageAssetResponse( id=str(f.get("id") or ""), filename=str(f.get("filename") or ""), content_type=f.get("content_type"), size_bytes=f.get("size_bytes"), status=str(f.get("status") or ""), created_at=f.get("created_at"), is_archive=_is_archive_file( filename=str(f.get("filename") or ""), content_type=f.get("content_type"), ), public_url=f.get("public_url"), ) for f in raw.get("files") or [] ] archives = sorted( [f for f in files if f.is_archive], key=lambda x: x.created_at or "", reverse=True, ) others = sorted( [f for f in files if not f.is_archive], key=lambda x: x.created_at or "", reverse=True, ) return StorageAssetsListResponse( bucket_name=bucket_name, files=archives + others, next_cursor=raw.get("next_cursor"), access_policy=settings.access_policy, cdn_base_url=settings.cdn_base_url, public_urls_enabled=settings.public_urls_enabled, ) async def get_bucket_settings( self, project: Project, *, bucket_name: str = DROPBOX_IMPORT_BUCKET, ) -> DropboxBucketSettingsResponse: await self._ensure_bucket(project.id, bucket_name) row = await self._bucket_repo.get_by_name(project.id, bucket_name) return _bucket_settings_from_row(bucket_name, row) async def update_bucket_settings( self, project: Project, body: UpdateDropboxBucketSettingsRequest, ) -> DropboxBucketSettingsResponse: await self._ensure_bucket(project.id, body.bucket_name) if body.public_urls_enabled and not body.cdn_base_url: raise ValueError("cdn_base_url_required_for_public_urls") access_policy = "public" if body.public_urls_enabled else "private" # Empty string clears cdn_custom_domain (update_bucket skips explicit None). cdn_domain: str | None if body.public_urls_enabled: cdn_domain = ((body.cdn_base_url or "")[:253] or None) else: cdn_domain = "" row = await self._bucket_repo.get_by_name(project.id, body.bucket_name) if row: updated = await self._storage_service.update_bucket( project.id, body.bucket_name, UpdateBucketRequest( access_policy=access_policy, cdn_enabled=body.public_urls_enabled, cdn_custom_domain=cdn_domain, ), ) else: updated = await self._storage_service.create_bucket( project.id, CreateBucketRequest( name=body.bucket_name, access_policy=access_policy, cdn_enabled=body.public_urls_enabled, cdn_custom_domain=(body.cdn_base_url or "")[:253] or None, ), ) return DropboxBucketSettingsResponse( bucket_name=body.bucket_name, access_policy=updated.access_policy, cdn_enabled=updated.cdn_enabled, cdn_base_url=updated.cdn_custom_domain, public_urls_enabled=bool( updated.access_policy == "public" and updated.cdn_enabled and updated.cdn_custom_domain ), ) async def get_asset_access_url( self, project: Project, file_id: str, *, bucket_name: str = DROPBOX_IMPORT_BUCKET, expires_in: int = 86400, ) -> AssetAccessUrlResponse: bucket_row = await self._bucket_repo.get_by_name(project.id, bucket_name) cloud_client = await get_cloud_client_for_workspace(project.workspace_id, self._db) stored = await cloud_client.get_storage_file( project_id=str(project.id), bucket_name=bucket_name, file_id=file_id, ) if not stored: raise ValueError("file_not_found") public_url = stored.get("public_url") if ( public_url and bucket_row and bucket_row.access_policy == "public" and bucket_row.cdn_enabled and bucket_row.cdn_custom_domain ): return AssetAccessUrlResponse( file_id=file_id, url=str(public_url), url_type="cdn", expires_in=None, ) signed = await cloud_client.presign_storage_file_download( project_id=str(project.id), bucket_name=bucket_name, file_id=file_id, expires_in=expires_in, ) if not signed.get("ok"): raise RuntimeError(signed.get("error") or "presign_failed") download_url = signed.get("download_url") if not download_url: raise RuntimeError("presign_missing_url") return AssetAccessUrlResponse( file_id=file_id, url=str(download_url), url_type="presigned", expires_in=int(signed.get("expires_in") or expires_in), ) async def delete_asset( self, project: Project, file_id: str, *, bucket_name: str = DROPBOX_IMPORT_BUCKET, ) -> DeleteStorageAssetResponse: cloud_client = await get_cloud_client_for_workspace(project.workspace_id, self._db) existing = await cloud_client.get_storage_file( project_id=str(project.id), bucket_name=bucket_name, file_id=file_id, ) if not existing: raise ValueError("file_not_found") result = await cloud_client.delete_storage_file( project_id=str(project.id), bucket_name=bucket_name, file_id=file_id, ) if not result.get("ok"): raise RuntimeError(result.get("error") or "cloud_delete_failed") return DeleteStorageAssetResponse(ok=True, file_id=file_id)