"""Durable, retrying delivery of outbound webhooks.

A workflow's final webhook must survive a transient network error. Rather than
firing the HTTP POST inline and forgetting it, ``run_integrations`` persists a
``WebhookDeliveryModel`` row and enqueues :func:`deliver_webhook`. This task sends
the request and, on a *transient* failure, schedules the next attempt with
exponential backoff -- up to ``max_attempts``, after which the delivery is parked
as ``dead_letter`` for inspection. Permanent failures (most 4xx) dead-letter
immediately instead of looping.

A periodic :func:`sweep_webhook_deliveries` cron re-enqueues any ``pending``
delivery whose attempt is overdue, so deliveries survive worker restarts / lost
ARQ jobs. The DB row is the source of truth; this task is idempotent and only
acts on a delivery that is still ``pending``.
"""

import json
import re
from datetime import UTC, datetime, timedelta
from typing import Any, Optional
from urllib.parse import urlsplit, urlunsplit

import httpx
from loguru import logger
from pipecat.utils.run_context import set_current_org_id, set_current_run_id

from api.constants import DEFAULT_WEBHOOK_DELIVERY_CONFIG
from api.db import db_client
from api.db.models import WebhookDeliveryModel
from api.errors.failure import (
    DograhFailure,
    ErrorSource,
    classify_exception,
    classify_http_response,
    log_failure,
)
from api.tasks.function_names import FunctionNames
from api.utils.credential_auth import build_auth_header

# HTTP statuses that are worth retrying even though the server answered.
_RETRYABLE_STATUS_CODES = {408, 425, 429, 500, 502, 503, 504}

_REDACTED = "[REDACTED]"
_MAX_REQUEST_LOG_CHARS = 8_000
_SENSITIVE_FIELD_RE = re.compile(
    r"(?:^|_)(?:"
    r"authorization|cookie|credential|password|passwd|secret|signature|token|"
    r"api_?key|access_?key|private_?key|client_?secret|"
    r"email|phone|telephone|mobile|address|date_?of_?birth|dob|ssn|"
    r"card(?:_?(?:number|no))?|cvv\d*|cvc\d*|callback_url|redirect_uri"
    r")(?:$|_)",
    re.IGNORECASE,
)


def _normalize_webhook_field_name(key: Any) -> str:
    """Normalize common field-name styles before checking for sensitive terms."""
    name = re.sub(r"(?<=[a-z0-9])(?=[A-Z])", "_", str(key))
    name = re.sub(r"(?<=[A-Z])(?=[A-Z][a-z])", "_", name)
    return re.sub(r"[^a-zA-Z0-9]+", "_", name).strip("_").lower()


def _redact_webhook_value(value: Any) -> Any:
    """Redact known sensitive fields while retaining payload shape for debugging."""
    if isinstance(value, dict):
        return {
            str(key): (
                _REDACTED
                if _SENSITIVE_FIELD_RE.search(_normalize_webhook_field_name(key))
                else _redact_webhook_value(nested)
            )
            for key, nested in value.items()
        }
    if isinstance(value, (list, tuple)):
        return [_redact_webhook_value(item) for item in value]
    if isinstance(value, str):
        try:
            parsed = urlsplit(value)
            if parsed.scheme.lower() in {"http", "https"} and parsed.hostname:
                return _REDACTED
        except ValueError:
            pass
    return value


def _safe_webhook_url(url: str) -> str:
    """Return only the URL origin so path-based credentials cannot reach logs."""
    try:
        parsed = urlsplit(url)
        hostname = parsed.hostname or ""
        if not parsed.scheme or not hostname:
            return _REDACTED
        if ":" in hostname:
            hostname = f"[{hostname}]"
        if parsed.port is not None:
            hostname = f"{hostname}:{parsed.port}"
        return urlunsplit((parsed.scheme, hostname, "", "", ""))
    except (TypeError, ValueError):
        return _REDACTED


def _log_webhook_request(
    delivery: WebhookDeliveryModel,
    *,
    method: str,
    attempt: int,
    headers: dict[str, str],
) -> None:
    """Log the frozen request shape with known sensitive values redacted."""
    safe_headers = {
        key: (
            value
            if key.lower()
            in {
                "content-type",
                "x-dograh-delivery-id",
                "x-dograh-workflow-run-id",
                "x-dograh-delivery-attempt",
            }
            else _REDACTED
        )
        for key, value in headers.items()
    }
    request_data = {
        "method": method,
        "url": _safe_webhook_url(delivery.endpoint_url),
        "headers": safe_headers,
        "payload": (
            _redact_webhook_value(delivery.payload)
            if method in ("POST", "PUT", "PATCH")
            else None
        ),
    }
    rendered = json.dumps(request_data, default=str, ensure_ascii=False)
    if len(rendered) > _MAX_REQUEST_LOG_CHARS:
        rendered = rendered[:_MAX_REQUEST_LOG_CHARS] + "...[TRUNCATED]"
    logger.info(
        f"Webhook '{delivery.webhook_name}' delivery {delivery.id} request "
        f"(attempt {attempt}): {rendered}"
    )


def _delivery_job_id(delivery_id: int, attempt_count: int) -> str:
    """Deterministic ARQ job id so duplicate enqueues (task re-enqueue + sweeper)
    collapse to one job instead of double-sending."""
    return f"webhook-delivery-{delivery_id}-{attempt_count}"


def _backoff_seconds(attempt: int) -> int:
    """Exponential backoff (capped) for the next attempt after `attempt` failures."""
    base = DEFAULT_WEBHOOK_DELIVERY_CONFIG["base_delay_seconds"]
    cap = DEFAULT_WEBHOOK_DELIVERY_CONFIG["max_delay_seconds"]
    return min(base * (2 ** (attempt - 1)), cap)


async def _enqueue_delivery(
    delivery_id: int,
    attempt_count: int,
    defer_by: int = 0,
    reclaim_token: Optional[int] = None,
):
    """Enqueue a delivery attempt with a dedup-safe job id.

    The normal (task self-retry) path uses a deterministic id so a retry and a
    sweeper pass for the *same* attempt collapse to one job. The sweeper passes a
    ``reclaim_token`` (the lease timestamp) to get a distinct id, so reconciling a
    delivered-but-unrecorded row is not deduped against the original attempt's
    already-completed job. The atomic claim still guarantees at most one send.
    """
    from api.tasks.arq import enqueue_job  # lazy import avoids circular import

    if reclaim_token is not None:
        job_id = f"webhook-delivery-reclaim-{delivery_id}-{reclaim_token}"
    else:
        job_id = _delivery_job_id(delivery_id, attempt_count)

    await enqueue_job(
        FunctionNames.DELIVER_WEBHOOK,
        delivery_id,
        _job_id=job_id,
        _defer_by=defer_by,
    )


async def _build_headers(delivery: WebhookDeliveryModel, attempt: int) -> dict:
    """Assemble request headers, re-resolving credential auth at send time so
    secrets are never persisted on the delivery row and rotation is honoured."""
    headers = {"Content-Type": "application/json"}

    if delivery.credential_uuid:
        credential = await db_client.get_credential_by_uuid(
            delivery.credential_uuid, delivery.organization_id
        )
        if credential:
            headers.update(build_auth_header(credential))
        else:
            logger.warning(
                f"Credential {delivery.credential_uuid} not found for webhook "
                f"'{delivery.webhook_name}' (delivery {delivery.id})"
            )

    for h in delivery.custom_headers or []:
        key, value = h.get("key"), h.get("value")
        if key and value:
            headers[key] = value

    # Stable idempotency signal so the receiver can dedupe retried deliveries.
    headers["X-Dograh-Delivery-Id"] = delivery.delivery_uuid
    headers["X-Dograh-Workflow-Run-Id"] = str(delivery.workflow_run_id)
    headers["X-Dograh-Delivery-Attempt"] = str(attempt)
    return headers


async def _handle_transient_failure(
    delivery: WebhookDeliveryModel,
    attempt: int,
    error: str,
    status_code: Optional[int],
) -> bool:
    """Schedule a backed-off retry, or dead-letter once attempts are exhausted."""
    if attempt >= delivery.max_attempts:
        await db_client.mark_webhook_delivery_dead_letter(
            delivery.id, attempt, error, status_code
        )
        return True

    delay = _backoff_seconds(attempt)
    scheduled_for = datetime.now(UTC) + timedelta(seconds=delay)
    await db_client.schedule_webhook_delivery_retry(
        delivery_id=delivery.id,
        attempt_count=attempt,
        scheduled_for=scheduled_for,
        last_error=error,
        last_status_code=status_code,
    )
    await _enqueue_delivery(delivery.id, attempt_count=attempt, defer_by=delay)
    logger.warning(
        f"Webhook '{delivery.webhook_name}' delivery {delivery.id} attempt {attempt} "
        f"failed ({error}); retrying in {delay}s "
        f"(attempt {attempt + 1}/{delivery.max_attempts})"
    )
    return False


def _log_dead_letter_failure(
    delivery: WebhookDeliveryModel, failure: DograhFailure
) -> None:
    log_failure(
        failure,
        organization_id=delivery.organization_id,
        workflow_run_id=delivery.workflow_run_id,
        delivery_id=delivery.id,
    )


async def deliver_webhook(_ctx, delivery_id: int) -> None:
    """Send one webhook delivery attempt and record the outcome.

    Concurrency-safe: the delivery is atomically *claimed* before the HTTP
    request (a conditional update only one worker can win), so a duplicate
    enqueue or sweeper re-injection cannot double-send. A claim that returns
    nothing means another worker owns it, or it is no longer pending/due -- a
    no-op.
    """
    # Lease long enough to outlast a full attempt so the sweeper does not reclaim
    # a delivery that is still in flight.
    lease_seconds = DEFAULT_WEBHOOK_DELIVERY_CONFIG["timeout_seconds"] + 60
    delivery = await db_client.claim_webhook_delivery(delivery_id, lease_seconds)
    if delivery is None:
        logger.debug(
            f"Webhook delivery {delivery_id} not claimable "
            f"(already claimed, not pending, or not yet due); skipping"
        )
        return

    set_current_run_id(str(delivery.workflow_run_id))
    set_current_org_id(delivery.organization_id)
    attempt = delivery.attempt_count + 1
    method = (delivery.http_method or "POST").upper()
    timeout = DEFAULT_WEBHOOK_DELIVERY_CONFIG["timeout_seconds"]

    try:
        headers = await _build_headers(delivery, attempt)
        _log_webhook_request(
            delivery,
            method=method,
            attempt=attempt,
            headers=headers,
        )

        async with httpx.AsyncClient() as client:
            if method in ("POST", "PUT", "PATCH"):
                response = await client.request(
                    method=method,
                    url=delivery.endpoint_url,
                    json=delivery.payload,
                    headers=headers,
                    timeout=timeout,
                )
            else:  # GET, DELETE
                response = await client.request(
                    method=method,
                    url=delivery.endpoint_url,
                    headers=headers,
                    timeout=timeout,
                )

        response.raise_for_status()
    except httpx.HTTPStatusError as e:
        status_code = e.response.status_code
        error = f"HTTP {status_code}: {e.response.text[:200]}"
        failure = classify_http_response(
            status_code,
            error,
            source=ErrorSource.WEBHOOK,
            provider="webhook",
            error_owner="user",
        )
        if status_code in _RETRYABLE_STATUS_CODES:
            dead_lettered = await _handle_transient_failure(
                delivery, attempt, error, status_code
            )
            if dead_lettered:
                _log_dead_letter_failure(delivery, failure)
        else:
            # Permanent (auth/validation/not-found): retrying won't help. Park it.
            await db_client.mark_webhook_delivery_dead_letter(
                delivery.id, attempt, error, status_code
            )
            _log_dead_letter_failure(delivery, failure)
        return
    except httpx.RequestError as e:
        # Connect/read timeouts, DNS, connection resets -- the transient class that
        # previously lost the webhook entirely. str(e) is often empty, so use repr.
        dead_lettered = await _handle_transient_failure(
            delivery, attempt, repr(e), None
        )
        if dead_lettered:
            _log_dead_letter_failure(
                delivery,
                classify_exception(
                    e,
                    source=ErrorSource.WEBHOOK,
                    provider="webhook",
                    error_owner="user",
                ),
            )
        return
    except Exception as e:
        # Unexpected (e.g. a bug): don't loop on it, surface as dead-letter.
        await db_client.mark_webhook_delivery_dead_letter(
            delivery.id, attempt, repr(e), None
        )
        _log_dead_letter_failure(
            delivery,
            classify_exception(
                e,
                source=ErrorSource.WEBHOOK,
                provider="webhook",
                error_owner="user",
            ),
        )
        return

    # The receiver accepted the payload (2xx). Recording success must NOT be able
    # to dead-letter an already-delivered webhook: if this DB write fails, log and
    # leave the row claimed-but-pending so the sweeper reconciles it once the
    # lease expires (the receiver dedups the re-send via X-Dograh-Delivery-Id).
    try:
        await db_client.mark_webhook_delivery_succeeded(
            delivery.id, attempt, response.status_code
        )
        logger.info(
            f"Webhook '{delivery.webhook_name}' delivery {delivery.id} succeeded: "
            f"{response.status_code} (attempt {attempt})"
        )
    except Exception as e:
        logger.error(
            f"Webhook '{delivery.webhook_name}' delivery {delivery.id} was "
            f"delivered ({response.status_code}) but recording success failed; "
            f"leaving it for the sweeper to reconcile after the lease expires: {e!r}"
        )


async def sweep_webhook_deliveries(_ctx) -> None:
    """Safety net: re-enqueue pending deliveries whose attempt is overdue.

    Handles ARQ jobs lost to a worker restart or Redis flush. Re-enqueuing uses the
    same deterministic job id, so if the original deferred job still exists this is a
    no-op; it only re-injects genuinely lost work. ``deliver_webhook`` is idempotent.
    """
    page_size = 100
    after_id = 0
    total = 0
    while True:
        # Re-enqueuing does not change a row's due state, so we cannot page by
        # re-querying the first rows (we'd loop on the same page). Page by id
        # instead to drain the whole backlog -- e.g. after a prolonged outage.
        due = await db_client.get_due_webhook_deliveries(
            now=datetime.now(UTC), limit=page_size, after_id=after_id
        )
        if not due:
            break
        for delivery in due:
            # A reclaim token (the current lease timestamp) gives this a fresh job
            # id so it is not deduped against the original attempt's completed job
            # -- otherwise a delivered-but-unrecorded row could sit until ARQ's
            # result retention clears.
            reclaim_token = (
                int(delivery.scheduled_for.timestamp()) if delivery.scheduled_for else 0
            )
            await _enqueue_delivery(
                delivery.id,
                attempt_count=delivery.attempt_count,
                reclaim_token=reclaim_token,
            )
        total += len(due)
        after_id = due[-1].id
        if len(due) < page_size:
            break

    if total:
        logger.info(f"Webhook delivery sweep: re-enqueued {total} due deliveries")