suna/backend/pipedream/app_service.py

from typing import List, Optional, Dict, Any
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
import os
import re
import httpx
import json
import asyncio
from utils.logger import logger

class AppSlug:
    def __init__(self, value: str):
        if not value or not isinstance(value, str):
            raise ValueError("AppSlug must be a non-empty string")
        if not re.match(r'^[a-z0-9_-]+$', value):
            raise ValueError("AppSlug must contain only lowercase letters, numbers, hyphens, and underscores")
        self.value = value

class SearchQuery:
    def __init__(self, value: Optional[str] = None):
        self.value = value
    
    def is_empty(self) -> bool:
        return not self.value or not self.value.strip()

class Category:
    def __init__(self, value: str):
        if not value or not isinstance(value, str):
            raise ValueError("Category must be a non-empty string")
        self.value = value

class PaginationCursor:
    def __init__(self, value: Optional[str] = None):
        self.value = value
    
    def has_more(self) -> bool:
        return self.value is not None

# Domain Entities
class AuthType(Enum):
    OAUTH = "oauth"
    API_KEY = "api_key"
    BASIC = "basic"
    NONE = "none"
    KEYS = "keys"
    CUSTOM = "custom"
    
    @classmethod
    def _missing_(cls, value):
        if isinstance(value, str):
            return cls.CUSTOM
        return super()._missing_(value)

@dataclass
class App:
    name: str
    slug: str
    description: str
    category: str
    logo_url: Optional[str] = None
    auth_type: AuthType = AuthType.OAUTH
    is_verified: bool = False
    url: Optional[str] = None
    tags: List[str] = field(default_factory=list)
    featured_weight: int = 0
    
    def is_featured(self) -> bool:
        return self.featured_weight > 0

class AppServiceError(Exception):
    pass

class AppNotFoundError(AppServiceError):
    pass

class InvalidAppSlugError(AppServiceError):
    pass

class AuthenticationError(AppServiceError):
    pass

class RateLimitError(AppServiceError):
    pass

class AppService:
    def __init__(self):
        self.base_url = "https://api.pipedream.com/v1"
        self.session: Optional[httpx.AsyncClient] = None
        self.access_token: Optional[str] = None
        self.token_expires_at: Optional[datetime] = None
        self._semaphore = asyncio.Semaphore(10)

    async def _get_session(self) -> httpx.AsyncClient:
        if self.session is None or self.session.is_closed:
            self.session = httpx.AsyncClient(
                timeout=httpx.Timeout(30.0),
                headers={"User-Agent": "Suna-Pipedream-Client/1.0"}
            )
        return self.session
    
    async def _ensure_access_token(self) -> str:
        if self.access_token and self.token_expires_at:
            if datetime.utcnow() < (self.token_expires_at - timedelta(minutes=5)):
                return self.access_token
        return await self._fetch_fresh_token()
    
    async def _fetch_fresh_token(self) -> str:
        project_id = os.getenv("PIPEDREAM_PROJECT_ID")
        client_id = os.getenv("PIPEDREAM_CLIENT_ID")
        client_secret = os.getenv("PIPEDREAM_CLIENT_SECRET")
        
        if not all([project_id, client_id, client_secret]):
            raise AuthenticationError("Missing required environment variables")
        
        session = await self._get_session()
        
        try:
            response = await session.post(
                f"{self.base_url}/oauth/token",
                data={
                    "grant_type": "client_credentials",
                    "client_id": client_id,
                    "client_secret": client_secret
                }
            )
            response.raise_for_status()
            
            data = response.json()
            self.access_token = data["access_token"]
            expires_in = data.get("expires_in", 3600)
            self.token_expires_at = datetime.utcnow() + timedelta(seconds=expires_in)
            
            return self.access_token
            
        except httpx.HTTPStatusError as e:
            if e.response.status_code == 429:
                raise RateLimitError()
            raise AuthenticationError(f"Failed to obtain access token: {e}")
    
    async def _make_request(self, url: str, headers: Dict[str, str] = None, params: Dict[str, Any] = None) -> Dict[str, Any]:
        session = await self._get_session()
        access_token = await self._ensure_access_token()
        
        request_headers = {
            "Authorization": f"Bearer {access_token}",
            "Content-Type": "application/json"
        }
        
        if headers:
            request_headers.update(headers)
        
        try:
            response = await session.get(url, headers=request_headers, params=params)
            response.raise_for_status()
            return response.json()
        except httpx.HTTPStatusError as e:
            if e.response.status_code == 429:
                raise RateLimitError()
            raise AppServiceError(f"HTTP request failed: {e}")

    async def _search(self, query: SearchQuery, category: Optional[Category] = None, 
                    page: int = 1, limit: int = 20, cursor: Optional[PaginationCursor] = None) -> Dict[str, Any]:
        url = f"{self.base_url}/apps"
        params = {}
        
        if not query.is_empty():
            params["q"] = query.value
        if category:
            params["category"] = category.value
        if cursor and cursor.value:
            params["after"] = cursor.value
        
        try:
            data = await self._make_request(url, params=params)
            apps = []
            for app_data in data.get("data", []):
                try:
                    app = self._map_to_domain(app_data)
                    apps.append(app)
                except Exception as e:
                    logger.warning(f"Error mapping app data: {str(e)}")
                    continue
            
            page_info = data.get("page_info", {})
            page_info["has_more"] = bool(page_info.get("end_cursor"))
            
            logger.info(f"Found {len(apps)} apps from search")
            
            return {
                "success": True,
                "apps": apps,
                "page_info": page_info,
                "total_count": page_info.get("total_count", 0)
            }
            
        except Exception as e:
            logger.error(f"Error searching apps: {str(e)}")
            return {
                "success": False,
                "error": str(e),
                "apps": [],
                "page_info": {},
                "total_count": 0
            }

    async def _get_by_slug(self, app_slug: str) -> Optional[App]:
        cache_key = f"pipedream:app:{app_slug}"
        try:
            from services import redis
            redis_client = await redis.get_client()
            cached_data = await redis_client.get(cache_key)
            
            if cached_data:
                logger.debug(f"Found cached app for slug: {app_slug}")
                cached_app_data = json.loads(cached_data)
                return self._map_cached_app_to_domain(cached_app_data)
        except Exception as e:
            logger.warning(f"Redis cache error for app {app_slug}: {e}")
        
        async with self._semaphore:
            url = f"{self.base_url}/apps"
            params = {"q": app_slug, "pageSize": 20}
            
            try:
                data = await self._make_request(url, params=params)
                
                apps = data.get("data", [])
                exact_match = next((app for app in apps if app.get("name_slug") == app_slug), None)
                
                if exact_match:
                    app = self._map_to_domain(exact_match)
                    
                    try:
                        from services import redis
                        redis_client = await redis.get_client()
                        app_data = self._map_domain_app_to_cache(app)
                        await redis_client.setex(cache_key, 21600, json.dumps(app_data))
                        logger.debug(f"Cached app: {app_slug}")
                    except Exception as e:
                        logger.warning(f"Failed to cache app {app_slug}: {e}")
                    
                    return app
                
                return None
                
            except Exception as e:
                logger.error(f"Error getting app by slug: {str(e)}")
                return None

    async def _get_popular(self, category: Optional[str] = None, limit: int = 100) -> List[App]:
        popular_slugs = [
            "slack", "microsoft_teams", "discord", "zoom", "telegram_bot_api",
            "gmail", "microsoft_outlook", "google_calendar", "microsoft_exchange", "calendly",
            "google_drive", "microsoft_onedrive", "dropbox", "google_docs", "google_sheets",
            "notion", "asana", "monday", "trello", "linear", "jira", "clickup",
            "salesforce", "hubspot", "pipedrive", "zendesk", "freshdesk", "intercom",
            "github", "gitlab", "bitbucket", "docker", "jenkins", "vercel", "netlify",
            "supabase", "firebase", "mongodb", "postgresql", "mysql", "redis", "airtable",
            "openai", "anthropic", "hugging_face", "replicate",
            "google_analytics", "facebook", "instagram", "twitter", "linkedin", "mailchimp",
            "stripe", "paypal", "quickbooks", "xero", "square",
            "aws", "google_cloud", "microsoft_azure", "digitalocean", "heroku",
            "shopify", "woocommerce", "magento", "bigcommerce"
        ]
        
        apps = []
        batch_size = 20
        target_slugs = popular_slugs[:limit]
        
        async def fetch_app(slug: str):
            try:
                app = await self._get_by_slug(slug)
                if app and (not category or app.category == category):
                    return app
                return None
            except Exception as e:
                logger.warning(f"Error fetching popular app {slug}: {e}")
                return None
        
        for i in range(0, len(target_slugs), batch_size):
            batch_slugs = target_slugs[i:i+batch_size]
            
            if i > 0:
                await asyncio.sleep(0.1)
            
            batch_tasks = [fetch_app(slug) for slug in batch_slugs]
            batch_results = await asyncio.gather(*batch_tasks, return_exceptions=True)
            
            for result in batch_results:
                if isinstance(result, App):
                    apps.append(result)
                    
                if len(apps) >= limit:
                    break
            
            if len(apps) >= limit:
                break
        
        return apps

    async def _get_by_category(self, category: str, limit: int = 20) -> List[App]:
        query = SearchQuery(None)
        category_obj = Category(category)
        result = await self._search(query, category_obj, limit=limit)
        return result.get("apps", [])

    def _map_to_domain(self, app_data: Dict[str, Any]) -> App:
        try:
            auth_type_str = app_data.get("auth_type", "oauth")
            auth_type = AuthType(auth_type_str)
        except ValueError:
            logger.warning(f"Unknown auth type '{auth_type_str}', using CUSTOM")
            auth_type = AuthType.CUSTOM
        
        return App(
            name=app_data.get("name", "Unknown"),
            slug=app_data.get("name_slug", ""),
            description=app_data.get("description", ""),
            category=app_data.get("category", "Other"),
            logo_url=app_data.get("img_src"),
            auth_type=auth_type,
            is_verified=app_data.get("verified", False),
            url=app_data.get("url"),
            tags=app_data.get("tags", []),
            featured_weight=app_data.get("featured_weight", 0)
        )
    
    def _map_domain_app_to_cache(self, app: App) -> Dict[str, Any]:
        return {
            "name": app.name,
            "name_slug": app.slug,
            "description": app.description,
            "category": app.category,
            "img_src": app.logo_url,
            "auth_type": app.auth_type.value,
            "verified": app.is_verified,
            "url": app.url,
            "tags": app.tags,
            "featured_weight": app.featured_weight
        }
    
    def _map_cached_app_to_domain(self, app_data: Dict[str, Any]) -> App:
        try:
            auth_type_str = app_data.get("auth_type", "oauth")
            auth_type = AuthType(auth_type_str)
        except ValueError:
            logger.warning(f"Unknown auth type '{auth_type_str}', using CUSTOM")
            auth_type = AuthType.CUSTOM
        
        return App(
            name=app_data.get("name", "Unknown"),
            slug=app_data.get("name_slug", ""),
            description=app_data.get("description", ""),
            category=app_data.get("category", "Other"),
            logo_url=app_data.get("img_src"),
            auth_type=auth_type,
            is_verified=app_data.get("verified", False),
            url=app_data.get("url"),
            tags=app_data.get("tags", []),
            featured_weight=app_data.get("featured_weight", 0)
        )

    async def search_apps(
        self,
        query: Optional[str] = None,
        category: Optional[str] = None,
        page: int = 1,
        limit: int = 20,
        cursor: Optional[str] = None
    ) -> Dict[str, Any]:
        search_query = SearchQuery(query)
        category_vo = Category(category) if category else None
        cursor_vo = PaginationCursor(cursor) if cursor else None
        
        logger.info(f"Searching apps: query='{query}', category='{category}', page={page}")
        
        result = await self._search(search_query, category_vo, page, limit, cursor_vo)
        
        logger.info(f"Found {len(result.get('apps', []))} apps")
        return result

    async def get_app_by_slug(self, app_slug: str) -> Optional[App]:
        logger.info(f"Getting app by slug: {app_slug}")
        
        app = await self._get_by_slug(app_slug)
        
        if app:
            logger.info(f"Found app: {app.name}")
        else:
            logger.info(f"App not found: {app_slug}")
        
        return app

    async def get_popular_apps(self, category: Optional[str] = None, limit: int = 10) -> List[App]:
        logger.info(f"Getting popular apps: category='{category}', limit={limit}")
        
        apps = await self._get_popular(category, limit)
        
        logger.info(f"Found {len(apps)} popular apps")
        return apps

    async def close(self):
        if self.session and not self.session.is_closed:
            await self.session.aclose()
    
    async def __aenter__(self):
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.close()


_app_service = None

def get_app_service() -> AppService:
    global _app_service
    if _app_service is None:
        _app_service = AppService()
    return _app_service


PipedreamException = AppServiceError
HttpClientException = AppServiceError
AuthenticationException = AuthenticationError
RateLimitException = RateLimitError