""" Storage Service - business logic for storage provider management. """ import json from typing import Optional, List, Dict, Any, Tuple, TYPE_CHECKING if TYPE_CHECKING: from database import Database from storage_providers import StorageProvidersModule STORAGE_PROVIDERS_INFO = { "pinata": {"name": "Pinata", "desc": "1GB free, IPFS pinning", "color": "blue"}, "web3storage": {"name": "web3.storage", "desc": "IPFS + Filecoin", "color": "green"}, "nftstorage": {"name": "NFT.Storage", "desc": "Free for NFTs", "color": "pink"}, "infura": {"name": "Infura IPFS", "desc": "5GB free", "color": "orange"}, "filebase": {"name": "Filebase", "desc": "5GB free, S3+IPFS", "color": "cyan"}, "storj": {"name": "Storj", "desc": "25GB free", "color": "indigo"}, "local": {"name": "Local Storage", "desc": "Your own disk", "color": "purple"}, } VALID_PROVIDER_TYPES = list(STORAGE_PROVIDERS_INFO.keys()) class StorageService: """Service for managing user storage providers.""" def __init__(self, database: "Database", storage_providers_module: "StorageProvidersModule") -> None: self.db = database self.providers = storage_providers_module async def list_storages(self, actor_id: str) -> List[Dict[str, Any]]: """List all storage providers for a user with usage stats.""" storages = await self.db.get_user_storage(actor_id) for storage in storages: usage = await self.db.get_storage_usage(storage["id"]) storage["used_bytes"] = usage["used_bytes"] storage["pin_count"] = usage["pin_count"] storage["donated_gb"] = storage["capacity_gb"] // 2 # Mask sensitive config keys for display if storage.get("config"): config = storage["config"] if isinstance(storage["config"], dict) else json.loads(storage["config"]) masked = {} for k, v in config.items(): if "key" in k.lower() or "token" in k.lower() or "secret" in k.lower(): masked[k] = v[:4] + "..." + v[-4:] if len(str(v)) > 8 else "****" else: masked[k] = v storage["config_display"] = masked return storages async def get_storage(self, storage_id: int, actor_id: str) -> Optional[Dict[str, Any]]: """Get a specific storage provider.""" storage = await self.db.get_storage_by_id(storage_id) if not storage: return None if storage["actor_id"] != actor_id: return None usage = await self.db.get_storage_usage(storage_id) storage["used_bytes"] = usage["used_bytes"] storage["pin_count"] = usage["pin_count"] storage["donated_gb"] = storage["capacity_gb"] // 2 return storage async def add_storage( self, actor_id: str, provider_type: str, config: Dict[str, Any], capacity_gb: int = 5, provider_name: Optional[str] = None, description: Optional[str] = None, ) -> Tuple[Optional[int], Optional[str]]: """Add a new storage provider. Returns (storage_id, error_message).""" if provider_type not in VALID_PROVIDER_TYPES: return None, f"Invalid provider type: {provider_type}" # Test connection before saving provider = self.providers.create_provider(provider_type, { **config, "capacity_gb": capacity_gb }) if not provider: return None, "Failed to create provider with given config" success, message = await provider.test_connection() if not success: return None, f"Provider connection failed: {message}" # Generate name if not provided if not provider_name: existing = await self.db.get_user_storage_by_type(actor_id, provider_type) provider_name = f"{provider_type}-{len(existing) + 1}" storage_id = await self.db.add_user_storage( actor_id=actor_id, provider_type=provider_type, provider_name=provider_name, config=config, capacity_gb=capacity_gb, description=description ) if not storage_id: return None, "Failed to save storage provider" return storage_id, None async def update_storage( self, storage_id: int, actor_id: str, config: Optional[Dict[str, Any]] = None, capacity_gb: Optional[int] = None, is_active: Optional[bool] = None, ) -> Tuple[bool, Optional[str]]: """Update a storage provider. Returns (success, error_message).""" storage = await self.db.get_storage_by_id(storage_id) if not storage: return False, "Storage provider not found" if storage["actor_id"] != actor_id: return False, "Not authorized" # Test new config if provided if config: existing_config = storage["config"] if isinstance(storage["config"], dict) else json.loads(storage["config"]) new_config = {**existing_config, **config} provider = self.providers.create_provider(storage["provider_type"], { **new_config, "capacity_gb": capacity_gb or storage["capacity_gb"] }) if provider: success, message = await provider.test_connection() if not success: return False, f"Provider connection failed: {message}" success = await self.db.update_user_storage( storage_id, config=config, capacity_gb=capacity_gb, is_active=is_active ) return success, None if success else "Failed to update storage provider" async def delete_storage(self, storage_id: int, actor_id: str) -> Tuple[bool, Optional[str]]: """Delete a storage provider. Returns (success, error_message).""" storage = await self.db.get_storage_by_id(storage_id) if not storage: return False, "Storage provider not found" if storage["actor_id"] != actor_id: return False, "Not authorized" success = await self.db.remove_user_storage(storage_id) return success, None if success else "Failed to remove storage provider" async def test_storage(self, storage_id: int, actor_id: str) -> Tuple[bool, str]: """Test storage provider connectivity. Returns (success, message).""" storage = await self.db.get_storage_by_id(storage_id) if not storage: return False, "Storage not found" if storage["actor_id"] != actor_id: return False, "Not authorized" config = storage["config"] if isinstance(storage["config"], dict) else json.loads(storage["config"]) provider = self.providers.create_provider(storage["provider_type"], { **config, "capacity_gb": storage["capacity_gb"] }) if not provider: return False, "Failed to create provider" return await provider.test_connection() async def list_by_type(self, actor_id: str, provider_type: str) -> List[Dict[str, Any]]: """List storage providers of a specific type.""" return await self.db.get_user_storage_by_type(actor_id, provider_type) def build_config_from_form(self, provider_type: str, form_data: Dict[str, Any]) -> Tuple[Optional[Dict[str, Any]], Optional[str]]: """Build provider config from form data. Returns (config, error).""" api_key = form_data.get("api_key") secret_key = form_data.get("secret_key") api_token = form_data.get("api_token") project_id = form_data.get("project_id") project_secret = form_data.get("project_secret") access_key = form_data.get("access_key") bucket = form_data.get("bucket") path = form_data.get("path") if provider_type == "pinata": if not api_key or not secret_key: return None, "Pinata requires API Key and Secret Key" return {"api_key": api_key, "secret_key": secret_key}, None elif provider_type == "web3storage": if not api_token: return None, "web3.storage requires API Token" return {"api_token": api_token}, None elif provider_type == "nftstorage": if not api_token: return None, "NFT.Storage requires API Token" return {"api_token": api_token}, None elif provider_type == "infura": if not project_id or not project_secret: return None, "Infura requires Project ID and Project Secret" return {"project_id": project_id, "project_secret": project_secret}, None elif provider_type == "filebase": if not access_key or not secret_key or not bucket: return None, "Filebase requires Access Key, Secret Key, and Bucket" return {"access_key": access_key, "secret_key": secret_key, "bucket": bucket}, None elif provider_type == "storj": if not access_key or not secret_key or not bucket: return None, "Storj requires Access Key, Secret Key, and Bucket" return {"access_key": access_key, "secret_key": secret_key, "bucket": bucket}, None elif provider_type == "local": if not path: return None, "Local storage requires a path" return {"path": path}, None return None, f"Unknown provider type: {provider_type}"