Triggers & Automation

def get_human_readable_schedule(cron_expression: str, timezone: str) -> str:
    # "0 9 * * *" -> "Daily at 9:00 AM"
    # "0 */6 * * *" -> "Every 6 hours"
    # "0 9 * * 1" -> "Every Monday at 9:00 AM"
    pass

# Verify webhook secret
secret = os.getenv("TRIGGER_WEBHOOK_SECRET")
if not secret:
    raise HTTPException(status_code=500, detail="Webhook secret not configured")

incoming_secret = request.headers.get("x-trigger-secret", "")
if not hmac.compare_digest(incoming_secret, secret):
    logger.warning(f"Invalid webhook secret for trigger {trigger_id}")
    raise HTTPException(status_code=401, detail="Unauthorized")

# Check trigger limits
if config.ENV_MODE != EnvMode.LOCAL:
    provider_trigger_type = await provider_service.get_provider_trigger_type(request.provider_id)
    trigger_type_str = 'scheduled' if provider_trigger_type.value == 'schedule' else 'app'
    
    from core.utils.limits_checker import check_trigger_limit
    limit_check = await check_trigger_limit(user_id, agent_id, trigger_type_str)
    
    if not limit_check['can_create']:
        error_detail = {
            "message": f"Maximum of {limit_check['limit']} {trigger_type_str} triggers allowed",
            "current_count": limit_check['current_count'],
            "limit": limit_check['limit'],
            "tier_name": limit_check['tier_name'],
            "trigger_type": trigger_type_str,
            "error_code": "TRIGGER_LIMIT_EXCEEDED"
        }
        raise HTTPException(status_code=402, detail=error_detail)

async def create_trigger(
    self,
    agent_id: str,
    provider_id: str,
    name: str,
    config: Dict[str, Any],
    description: Optional[str] = None
) -> Trigger:
    trigger_id = str(uuid.uuid4())
    
    # Validate config with provider
    provider_service = get_provider_service(self._db)
    validated_config = await provider_service.validate_trigger_config(provider_id, config)
    
    trigger_type = await provider_service.get_provider_trigger_type(provider_id)
    
    trigger = Trigger(
        trigger_id=trigger_id,
        agent_id=agent_id,
        provider_id=provider_id,
        trigger_type=trigger_type,
        name=name,
        description=description,
        is_active=True,
        config=validated_config,
        created_at=now,
        updated_at=now
    )
    
    # Setup with provider (e.g., register cron job)
    setup_success = await provider_service.setup_trigger(trigger)
    if not setup_success:
        raise ValueError(f"Failed to setup trigger with provider: {provider_id}")
    
    await self._save_trigger(trigger)
    
    return trigger

async def sync_triggers_to_version_config(agent_id: str):
    # Get current agent version
    agent_result = await client.table('agents')
        .select('current_version_id')
        .eq('agent_id', agent_id)
        .single()
        .execute()
    
    current_version_id = agent_result.data['current_version_id']
    
    # Get all triggers for this agent
    triggers_result = await client.table('agent_triggers')
        .select('*')
        .eq('agent_id', agent_id)
        .execute()
    
    triggers = triggers_result.data or []
    
    # Update version config with triggers
    version_result = await client.table('agent_versions')
        .select('config')
        .eq('version_id', current_version_id)
        .single()
        .execute()
    
    config = version_result.data.get('config', {})
    config['triggers'] = triggers
    
    await client.table('agent_versions')
        .update({'config': config})
        .eq('version_id', current_version_id)
        .execute()
    
    # Invalidate cache
    from core.cache.runtime_cache import invalidate_agent_config_cache
    await invalidate_agent_config_cache(agent_id)