src/crud/webhook.py

from logging import getLogger

from sqlalchemy import Select, select
from sqlalchemy.ext.asyncio import AsyncSession

from src import models, schemas
from src.config import settings
from src.crud.workspace import get_workspace
from src.exceptions import ResourceNotFoundException
from src.utils.types import GetOrCreateResult

logger = getLogger(__name__)


async def get_or_create_webhook_endpoint(
    db: AsyncSession,
    workspace_name: str,
    webhook: schemas.WebhookEndpointCreate,
) -> GetOrCreateResult[schemas.WebhookEndpoint]:
    """
    Get an existing webhook endpoint for a workspace or create it if missing.

    Args:
        db: Database session
        workspace_name: Name of the workspace
        webhook: Webhook endpoint creation schema

    Returns:
        GetOrCreateResult containing the webhook endpoint and whether it was created

    Raises:
        ResourceNotFoundException: If the workspace does not exist
        ValueError: If the workspace already has the maximum number of webhook
            endpoints
    """
    # Verify workspace exists
    await get_workspace(db, workspace_name=workspace_name)

    stmt = select(models.WebhookEndpoint).where(
        models.WebhookEndpoint.workspace_name == workspace_name,
    )
    result = await db.execute(stmt)
    endpoints = result.scalars().all()

    # Check if webhook already exists for this workspace
    for endpoint in endpoints:
        if endpoint.url == webhook.url:
            return GetOrCreateResult(
                schemas.WebhookEndpoint.model_validate(endpoint), created=False
            )

    # No more than WORKSPACE_LIMIT webhooks per workspace
    if len(endpoints) >= settings.WEBHOOK.MAX_WORKSPACE_LIMIT:
        raise ValueError(
            f"Maximum number of webhook endpoints ({settings.WEBHOOK.MAX_WORKSPACE_LIMIT}) reached for this workspace."
        )

    # Create new webhook endpoint
    webhook_endpoint = models.WebhookEndpoint(
        workspace_name=workspace_name,
        url=webhook.url,
    )
    db.add(webhook_endpoint)
    await db.commit()

    logger.debug("Webhook endpoint created: %s", webhook.url)
    return GetOrCreateResult(
        schemas.WebhookEndpoint.model_validate(webhook_endpoint), created=True
    )


async def list_webhook_endpoints(
    workspace_name: str,
) -> Select[tuple[models.WebhookEndpoint]]:
    """
    List all webhook endpoints, optionally filtered by workspace.

    Args:
        db: Database session
        workspace_name: Name of the workspace (optional)

    Returns:
        List of webhook endpoints
    """
    return select(models.WebhookEndpoint).where(
        models.WebhookEndpoint.workspace_name == workspace_name
    )


async def delete_webhook_endpoint(
    db: AsyncSession, workspace_name: str, endpoint_id: str
) -> None:
    """
    Delete a webhook endpoint.

    Args:
        db: Database session
        endpoint_id: ID of the webhook endpoint

    Raises:
        ResourceNotFoundException: If the webhook endpoint is not found
    """
    # Verify webhook endpoint exists
    stmt = select(models.WebhookEndpoint).where(
        models.WebhookEndpoint.id == endpoint_id,
        models.WebhookEndpoint.workspace_name == workspace_name,
    )
    result = await db.execute(stmt)
    endpoint = result.scalar_one_or_none()

    if not endpoint:
        raise ResourceNotFoundException(
            f"Webhook endpoint {endpoint_id} not found for workspace {workspace_name}"
        )

    await db.delete(endpoint)
    await db.commit()

    logger.debug("Webhook endpoint %s deleted", endpoint_id)