# This file is auto-generated by oagen. Do not edit.

from __future__ import annotations

from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union

if TYPE_CHECKING:
    from .._client import AsyncWorkOSClient, WorkOSClient

from .._types import RequestOptions, enum_value
from .models import WebhookEndpoint
from workos.common.models.create_webhook_endpoint_events import (
    CreateWebhookEndpointEvents,
)
from workos.common.models.pagination_order import PaginationOrder
from workos.common.models.update_webhook_endpoint_events import (
    UpdateWebhookEndpointEvents,
)
from workos.common.models.update_webhook_endpoint_status import (
    UpdateWebhookEndpointStatus,
)
from .._pagination import AsyncPage, SyncPage
import hashlib
import hmac
import json
import time

# @oagen-ignore-start
if TYPE_CHECKING:
    from workos.events.models import EventSchemaVariant
    # @oagen-ignore-end


class Webhooks:
    """Webhooks API resources."""

    def __init__(self, client: "WorkOSClient") -> None:
        self._client = client

    def list_webhook_endpoints(
        self,
        *,
        limit: Optional[int] = None,
        before: Optional[str] = None,
        after: Optional[str] = None,
        order: Optional[Union[PaginationOrder, str]] = "desc",
        request_options: Optional[RequestOptions] = None,
    ) -> SyncPage[WebhookEndpoint]:
        """List Webhook Endpoints

        Get a list of all of your existing webhook endpoints.

        Args:
            limit: Upper limit on the number of objects to return, between `1` and `100`. Defaults to `10`.
            before: An object ID that defines your place in the list. When the ID is not present, you are at the end of the list. For example, if you make a list request and receive 100 objects, ending with `"obj_123"`, your subsequent call can include `before="obj_123"` to fetch a new batch of objects before `"obj_123"`.
            after: An object ID that defines your place in the list. When the ID is not present, you are at the end of the list. For example, if you make a list request and receive 100 objects, ending with `"obj_123"`, your subsequent call can include `after="obj_123"` to fetch a new batch of objects after `"obj_123"`.
            order: Order the results by the creation time. Supported values are `"asc"` (ascending), `"desc"` (descending), and `"normal"` (descending with reversed cursor semantics where `before` fetches older records and `after` fetches newer records). Defaults to `desc`.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            SyncPage[WebhookEndpoint]

        Raises:
            AuthenticationError: If the API key is invalid (401).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        params = {
            k: v
            for k, v in {
                "limit": limit,
                "before": before,
                "after": after,
                "order": enum_value(order) if order is not None else None,
            }.items()
            if v is not None
        }
        return self._client.request_page(
            method="get",
            path=("webhook_endpoints",),
            model=WebhookEndpoint,
            params=params,
            request_options=request_options,
        )

    def create_webhook_endpoint(
        self,
        *,
        endpoint_url: str,
        events: List[Union[CreateWebhookEndpointEvents, str]],
        request_options: Optional[RequestOptions] = None,
    ) -> WebhookEndpoint:
        """Create a Webhook Endpoint

        Create a new webhook endpoint to receive event notifications.

        Args:
            endpoint_url: The HTTPS URL where webhooks will be sent.
            events: The events that the Webhook Endpoint is subscribed to.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            WebhookEndpoint

        Raises:
            ConflictError: If a conflict occurs (409).
            UnprocessableEntityError: If the request data is unprocessable (422).
            AuthenticationError: If the API key is invalid (401).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        body: Dict[str, Any] = {
            "endpoint_url": endpoint_url,
            "events": events,
        }
        return self._client.request(
            method="post",
            path=("webhook_endpoints",),
            body=body,
            model=WebhookEndpoint,
            request_options=request_options,
        )

    def update_webhook_endpoint(
        self,
        id: str,
        *,
        endpoint_url: Optional[str] = None,
        status: Optional[Union[UpdateWebhookEndpointStatus, str]] = None,
        events: Optional[List[Union[UpdateWebhookEndpointEvents, str]]] = None,
        request_options: Optional[RequestOptions] = None,
    ) -> WebhookEndpoint:
        """Update a Webhook Endpoint

        Update the properties of an existing webhook endpoint.

        Args:
            id: Unique identifier of the Webhook Endpoint.
            endpoint_url: The HTTPS URL where webhooks will be sent.
            status: Whether the Webhook Endpoint is enabled or disabled.
            events: The events that the Webhook Endpoint is subscribed to.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            WebhookEndpoint

        Raises:
            NotFoundError: If the resource is not found (404).
            ConflictError: If a conflict occurs (409).
            UnprocessableEntityError: If the request data is unprocessable (422).
            AuthenticationError: If the API key is invalid (401).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        body: Dict[str, Any] = {
            k: v
            for k, v in {
                "endpoint_url": endpoint_url,
                "status": enum_value(status) if status is not None else None,
                "events": events,
            }.items()
            if v is not None
        }
        return self._client.request(
            method="patch",
            path=("webhook_endpoints", str(id)),
            body=body,
            model=WebhookEndpoint,
            request_options=request_options,
        )

    def delete_webhook_endpoint(
        self,
        id: str,
        *,
        request_options: Optional[RequestOptions] = None,
    ) -> None:
        """Delete a Webhook Endpoint

        Delete an existing webhook endpoint.

        Args:
            id: Unique identifier of the Webhook Endpoint.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Raises:
            NotFoundError: If the resource is not found (404).
            AuthenticationError: If the API key is invalid (401).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        self._client.request(
            method="delete",
            path=("webhook_endpoints", str(id)),
            request_options=request_options,
        )

    # @oagen-ignore-start
    DEFAULT_TOLERANCE = 180

    def verify_event(
        self,
        *,
        event_body: Union[bytes, str],
        event_signature: str,
        secret: str,
        tolerance: Optional[int] = None,
    ) -> "EventSchemaVariant":
        """Verify and deserialize the signature of a Webhook event.

        Args:
            event_body: The raw Webhook request body (bytes or str).
            event_signature: The signature from the 'WorkOS-Signature' header.
            secret: The webhook endpoint secret from the WorkOS dashboard.
            tolerance: Maximum age of the event in seconds. Defaults to 180.

        Returns:
            EventSchemaVariant: The deserialized webhook event.

        Raises:
            ValueError: If the signature is invalid or the event is too old.
        """
        from workos.events.models import EventSchema

        self.verify_header(
            event_body=event_body,
            event_signature=event_signature,
            secret=secret,
            tolerance=tolerance,
        )
        body = event_body if isinstance(event_body, (str, bytes)) else str(event_body)
        return EventSchema.from_dict(json.loads(body))

    def verify_header(
        self,
        *,
        event_body: Union[bytes, str],
        event_signature: str,
        secret: str,
        tolerance: Optional[int] = None,
    ) -> None:
        """Verify the signature of a Webhook, raise ValueError if invalid.

        Args:
            event_body: The raw Webhook request body (bytes or str).
            event_signature: The signature from the 'WorkOS-Signature' header.
            secret: The webhook endpoint secret from the WorkOS dashboard.
            tolerance: Maximum age of the event in seconds. Defaults to 180.

        Raises:
            ValueError: If the signature cannot be verified or the event is too old.
        """
        try:
            issued_timestamp, signature_hash = event_signature.split(", ")
        except (ValueError, AttributeError) as exc:
            raise ValueError(
                "Unable to extract timestamp and signature hash from header",
                event_signature,
            ) from exc

        issued_timestamp = issued_timestamp[2:]
        signature_hash = signature_hash[3:]
        max_seconds_since_issued = (
            tolerance if tolerance is not None else self.DEFAULT_TOLERANCE
        )
        current_time = time.time()
        timestamp_in_seconds = int(issued_timestamp) / 1000
        seconds_since_issued = current_time - timestamp_in_seconds

        if abs(seconds_since_issued) > max_seconds_since_issued:
            raise ValueError("Timestamp outside the tolerance zone")

        body_str = (
            event_body.decode("utf-8") if isinstance(event_body, bytes) else event_body
        )
        unhashed_string = f"{issued_timestamp}.{body_str}"
        expected_signature = hmac.new(
            secret.encode("utf-8"),
            unhashed_string.encode("utf-8"),
            digestmod=hashlib.sha256,
        ).hexdigest()

        if not hmac.compare_digest(signature_hash, expected_signature):
            raise ValueError(
                "Signature hash does not match the expected signature hash for payload"
            )

    # @oagen-ignore-end


class AsyncWebhooks:
    """Webhooks API resources (async)."""

    def __init__(self, client: "AsyncWorkOSClient") -> None:
        self._client = client

    async def list_webhook_endpoints(
        self,
        *,
        limit: Optional[int] = None,
        before: Optional[str] = None,
        after: Optional[str] = None,
        order: Optional[Union[PaginationOrder, str]] = "desc",
        request_options: Optional[RequestOptions] = None,
    ) -> AsyncPage[WebhookEndpoint]:
        """List Webhook Endpoints

        Get a list of all of your existing webhook endpoints.

        Args:
            limit: Upper limit on the number of objects to return, between `1` and `100`. Defaults to `10`.
            before: An object ID that defines your place in the list. When the ID is not present, you are at the end of the list. For example, if you make a list request and receive 100 objects, ending with `"obj_123"`, your subsequent call can include `before="obj_123"` to fetch a new batch of objects before `"obj_123"`.
            after: An object ID that defines your place in the list. When the ID is not present, you are at the end of the list. For example, if you make a list request and receive 100 objects, ending with `"obj_123"`, your subsequent call can include `after="obj_123"` to fetch a new batch of objects after `"obj_123"`.
            order: Order the results by the creation time. Supported values are `"asc"` (ascending), `"desc"` (descending), and `"normal"` (descending with reversed cursor semantics where `before` fetches older records and `after` fetches newer records). Defaults to `desc`.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            AsyncPage[WebhookEndpoint]

        Raises:
            AuthenticationError: If the API key is invalid (401).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        params = {
            k: v
            for k, v in {
                "limit": limit,
                "before": before,
                "after": after,
                "order": enum_value(order) if order is not None else None,
            }.items()
            if v is not None
        }
        return await self._client.request_page(
            method="get",
            path=("webhook_endpoints",),
            model=WebhookEndpoint,
            params=params,
            request_options=request_options,
        )

    async def create_webhook_endpoint(
        self,
        *,
        endpoint_url: str,
        events: List[Union[CreateWebhookEndpointEvents, str]],
        request_options: Optional[RequestOptions] = None,
    ) -> WebhookEndpoint:
        """Create a Webhook Endpoint

        Create a new webhook endpoint to receive event notifications.

        Args:
            endpoint_url: The HTTPS URL where webhooks will be sent.
            events: The events that the Webhook Endpoint is subscribed to.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            WebhookEndpoint

        Raises:
            ConflictError: If a conflict occurs (409).
            UnprocessableEntityError: If the request data is unprocessable (422).
            AuthenticationError: If the API key is invalid (401).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        body: Dict[str, Any] = {
            "endpoint_url": endpoint_url,
            "events": events,
        }
        return await self._client.request(
            method="post",
            path=("webhook_endpoints",),
            body=body,
            model=WebhookEndpoint,
            request_options=request_options,
        )

    async def update_webhook_endpoint(
        self,
        id: str,
        *,
        endpoint_url: Optional[str] = None,
        status: Optional[Union[UpdateWebhookEndpointStatus, str]] = None,
        events: Optional[List[Union[UpdateWebhookEndpointEvents, str]]] = None,
        request_options: Optional[RequestOptions] = None,
    ) -> WebhookEndpoint:
        """Update a Webhook Endpoint

        Update the properties of an existing webhook endpoint.

        Args:
            id: Unique identifier of the Webhook Endpoint.
            endpoint_url: The HTTPS URL where webhooks will be sent.
            status: Whether the Webhook Endpoint is enabled or disabled.
            events: The events that the Webhook Endpoint is subscribed to.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            WebhookEndpoint

        Raises:
            NotFoundError: If the resource is not found (404).
            ConflictError: If a conflict occurs (409).
            UnprocessableEntityError: If the request data is unprocessable (422).
            AuthenticationError: If the API key is invalid (401).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        body: Dict[str, Any] = {
            k: v
            for k, v in {
                "endpoint_url": endpoint_url,
                "status": enum_value(status) if status is not None else None,
                "events": events,
            }.items()
            if v is not None
        }
        return await self._client.request(
            method="patch",
            path=("webhook_endpoints", str(id)),
            body=body,
            model=WebhookEndpoint,
            request_options=request_options,
        )

    async def delete_webhook_endpoint(
        self,
        id: str,
        *,
        request_options: Optional[RequestOptions] = None,
    ) -> None:
        """Delete a Webhook Endpoint

        Delete an existing webhook endpoint.

        Args:
            id: Unique identifier of the Webhook Endpoint.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Raises:
            NotFoundError: If the resource is not found (404).
            AuthenticationError: If the API key is invalid (401).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        await self._client.request(
            method="delete",
            path=("webhook_endpoints", str(id)),
            request_options=request_options,
        )

    # @oagen-ignore-start
    DEFAULT_TOLERANCE = 180

    def verify_event(
        self,
        *,
        event_body: Union[bytes, str],
        event_signature: str,
        secret: str,
        tolerance: Optional[int] = None,
    ) -> "EventSchemaVariant":
        """Verify and deserialize the signature of a Webhook event.

        Args:
            event_body: The raw Webhook request body (bytes or str).
            event_signature: The signature from the 'WorkOS-Signature' header.
            secret: The webhook endpoint secret from the WorkOS dashboard.
            tolerance: Maximum age of the event in seconds. Defaults to 180.

        Returns:
            EventSchemaVariant: The deserialized webhook event.

        Raises:
            ValueError: If the signature is invalid or the event is too old.
        """
        from workos.events.models import EventSchema

        self.verify_header(
            event_body=event_body,
            event_signature=event_signature,
            secret=secret,
            tolerance=tolerance,
        )
        body = event_body if isinstance(event_body, (str, bytes)) else str(event_body)
        return EventSchema.from_dict(json.loads(body))

    def verify_header(
        self,
        *,
        event_body: Union[bytes, str],
        event_signature: str,
        secret: str,
        tolerance: Optional[int] = None,
    ) -> None:
        """Verify the signature of a Webhook, raise ValueError if invalid.

        Args:
            event_body: The raw Webhook request body (bytes or str).
            event_signature: The signature from the 'WorkOS-Signature' header.
            secret: The webhook endpoint secret from the WorkOS dashboard.
            tolerance: Maximum age of the event in seconds. Defaults to 180.

        Raises:
            ValueError: If the signature cannot be verified or the event is too old.
        """
        try:
            issued_timestamp, signature_hash = event_signature.split(", ")
        except (ValueError, AttributeError) as exc:
            raise ValueError(
                "Unable to extract timestamp and signature hash from header",
                event_signature,
            ) from exc

        issued_timestamp = issued_timestamp[2:]
        signature_hash = signature_hash[3:]
        max_seconds_since_issued = (
            tolerance if tolerance is not None else self.DEFAULT_TOLERANCE
        )
        current_time = time.time()
        timestamp_in_seconds = int(issued_timestamp) / 1000
        seconds_since_issued = current_time - timestamp_in_seconds

        if abs(seconds_since_issued) > max_seconds_since_issued:
            raise ValueError("Timestamp outside the tolerance zone")

        body_str = (
            event_body.decode("utf-8") if isinstance(event_body, bytes) else event_body
        )
        unhashed_string = f"{issued_timestamp}.{body_str}"
        expected_signature = hmac.new(
            secret.encode("utf-8"),
            unhashed_string.encode("utf-8"),
            digestmod=hashlib.sha256,
        ).hexdigest()

        if not hmac.compare_digest(signature_hash, expected_signature):
            raise ValueError(
                "Signature hash does not match the expected signature hash for payload"
            )

    # @oagen-ignore-end
