"""Reliable webhook delivery service.

Events are persisted to the `webhook_events` table before delivery is attempted.
A background worker processes pending events with exponential backoff.
After max_attempts (default 5), events are marked as dead_letter.

Delivery statuses:
- pending: awaiting delivery attempt
- delivered: successfully delivered (2xx response)
- dead_letter: all attempts exhausted

Retry schedule (exponential backoff with jitter):
  attempt 1: immediate
  attempt 2: ~30s
  attempt 3: ~2min
  attempt 4: ~8min
  attempt 5: ~30min
"""

from __future__ import annotations

import asyncio
import random
import uuid
from datetime import datetime, timedelta, timezone
from typing import Any

import structlog
from sqlalchemy import delete, select
from sqlalchemy.ext.asyncio import AsyncSession

from server.db.engine import get_session_factory
from server.db.tables import WebhookEventRow

logger = structlog.stdlib.get_logger()

_webhook_url: str | None = None
_timeout: float = 5.0
_worker_task: asyncio.Task | None = None
_poll_interval: float = 5.0  # seconds between queue checks
# Event-type allowlist. An empty set means "no filter — deliver every
# event" (the backward-compatible default). When non-empty, only events
# whose type is in the set are enqueued.
_event_filter: frozenset[str] = frozenset()


def configure(
    url: str | None,
    timeout: float = 5.0,
    events: list[str] | None = None,
) -> None:
    """Set the global webhook URL, timeout, and event filter.

    Called at app startup with values from `Settings`. `events` is the
    event-type allowlist (`STATEWAVE_WEBHOOK_EVENTS`): an empty or None
    value disables filtering so every event is delivered.
    """
    global _webhook_url, _timeout, _event_filter
    _webhook_url = url
    _timeout = timeout
    _event_filter = frozenset(events) if events else frozenset()


def event_filter() -> frozenset[str]:
    """Return the active event-type allowlist.

    An empty set means no filter is configured — every event type is
    delivered. Exposed for tests and operator introspection.
    """
    return _event_filter


async def start_worker() -> None:
    """Start the background delivery worker."""
    global _worker_task
    if not _webhook_url:
        logger.info("webhook_worker_disabled", reason="no webhook URL configured")
        return
    _worker_task = asyncio.create_task(_delivery_loop())
    logger.info("webhook_worker_started", url=_webhook_url)


async def stop_worker() -> None:
    """Stop the background delivery worker."""
    global _worker_task
    if _worker_task:
        _worker_task.cancel()
        try:
            await _worker_task
        except asyncio.CancelledError:
            pass
        _worker_task = None
        logger.info("webhook_worker_stopped")


async def fire(
    event: str,
    payload: dict[str, Any],
    db: AsyncSession | None = None,
    *,
    tenant_id: str | None = None,
) -> uuid.UUID | None:
    """Enqueue a webhook event for delivery.

    If no webhook URL is configured, returns None (no-op). If an event
    filter is configured and `event` is not in the allowlist, the event
    is dropped before it reaches the delivery queue and None is
    returned. Otherwise the event is persisted immediately and delivered
    asynchronously.
    """
    if not _webhook_url:
        return None

    # Event-type allowlist (STATEWAVE_WEBHOOK_EVENTS). An empty filter
    # means "deliver everything"; a filtered-out event is never enqueued,
    # so it costs nothing in storage or delivery attempts.
    if _event_filter and event not in _event_filter:
        logger.debug("webhook_event_filtered", event_type=event)
        return None

    event_id = uuid.uuid4()
    now = datetime.now(timezone.utc)

    row = WebhookEventRow(
        id=event_id,
        tenant_id=tenant_id,
        event=event,
        payload={
            "event": event,
            "timestamp": now.isoformat(),
            "data": payload,
        },
        status="pending",
        attempts=0,
        next_attempt_at=now,  # deliver immediately
        created_at=now,
    )

    if db:
        db.add(row)
        # Will be committed with the caller's transaction
    else:
        async with get_session_factory()() as session:
            session.add(row)
            await session.commit()

    return event_id


def _backoff_seconds(attempt: int) -> float:
    """Exponential backoff with jitter: ~30s, ~2m, ~8m, ~30m."""
    base = 30 * (4 ** (attempt - 1))  # 30, 120, 480, 1920
    jitter = random.uniform(0.5, 1.5)
    return base * jitter


async def _delivery_loop() -> None:
    """Background loop that processes pending webhook events."""
    while True:
        try:
            await _process_pending_batch()
        except asyncio.CancelledError:
            raise
        except Exception:
            logger.error("webhook_worker_error", exc_info=True)
        await asyncio.sleep(_poll_interval)


async def _process_pending_batch(batch_size: int = 20) -> int:
    """Process a batch of pending events. Returns count processed."""
    now = datetime.now(timezone.utc)
    processed = 0

    async with get_session_factory()() as session:
        # Fetch pending events that are due for delivery
        stmt = (
            select(WebhookEventRow)
            .where(WebhookEventRow.status == "pending")
            .where(WebhookEventRow.next_attempt_at <= now)
            .order_by(WebhookEventRow.next_attempt_at)
            .limit(batch_size)
        )
        result = await session.execute(stmt)
        events = result.scalars().all()

        for event in events:
            await _attempt_delivery(event, session)
            processed += 1

        await session.commit()

    return processed


async def _attempt_delivery(event: WebhookEventRow, session: AsyncSession) -> None:
    """Attempt to deliver a single webhook event."""
    import httpx

    now = datetime.now(timezone.utc)
    event.attempts += 1
    event.last_attempt_at = now

    try:
        async with httpx.AsyncClient(timeout=_timeout) as client:
            resp = await client.post(_webhook_url, json=event.payload)  # type: ignore[arg-type]

        event.http_status = resp.status_code

        if 200 <= resp.status_code < 300:
            event.status = "delivered"
            event.delivered_at = now
            logger.debug(
                "webhook_delivered",
                event_id=str(event.id),
                event_type=event.event,
                attempt=event.attempts,
                status=resp.status_code,
            )
        elif _is_retryable_status(resp.status_code):
            _handle_failure(event, f"HTTP {resp.status_code}: {resp.text[:200]}")
        else:
            _handle_permanent_failure(
                event, f"HTTP {resp.status_code}: {resp.text[:200]}"
            )

    except Exception as exc:
        event.http_status = None
        _handle_failure(event, f"{type(exc).__name__}: {str(exc)[:200]}")


def _is_retryable_status(status_code: int) -> bool:
    """Return True when a later delivery attempt may reasonably succeed."""
    return status_code in {408, 409, 425, 429} or status_code >= 500


def _handle_permanent_failure(event: WebhookEventRow, error: str) -> None:
    """Dead-letter non-retryable delivery failures immediately."""
    event.last_error = error
    event.status = "dead_letter"
    event.next_attempt_at = None
    logger.warning(
        "webhook_permanent_failure",
        event_id=str(event.id),
        event_type=event.event,
        attempts=event.attempts,
        last_error=error,
    )


def _handle_failure(event: WebhookEventRow, error: str) -> None:
    """Handle a failed delivery attempt — retry or dead-letter."""
    event.last_error = error

    if event.attempts >= event.max_attempts:
        event.status = "dead_letter"
        logger.warning(
            "webhook_dead_letter",
            event_id=str(event.id),
            event_type=event.event,
            attempts=event.attempts,
            last_error=error,
        )
    else:
        # Schedule retry with backoff
        backoff = _backoff_seconds(event.attempts)
        event.next_attempt_at = datetime.now(timezone.utc) + timedelta(seconds=backoff)
        logger.info(
            "webhook_retry_scheduled",
            event_id=str(event.id),
            event_type=event.event,
            attempt=event.attempts,
            next_in_seconds=round(backoff),
        )


# ── Query helpers (for admin endpoints) ───────────────────────────────────


async def get_event_status(
    event_id: uuid.UUID, *, tenant_id: str | None = None
) -> dict[str, Any] | None:
    """Get the delivery status of a single webhook event.

    When ``tenant_id`` is provided, an event belonging to a different tenant is
    reported as not-found, so the operator's optional tenant filter behaves
    consistently with the list/purge endpoints.
    """
    async with get_session_factory()() as session:
        row = await session.get(WebhookEventRow, event_id)
        if not row:
            return None
        if tenant_id is not None and row.tenant_id != tenant_id:
            return None
        return {
            "id": str(row.id),
            "event": row.event,
            "status": row.status,
            "attempts": row.attempts,
            "max_attempts": row.max_attempts,
            "last_attempt_at": row.last_attempt_at.isoformat() if row.last_attempt_at else None,
            "next_attempt_at": row.next_attempt_at.isoformat() if row.next_attempt_at else None,
            "last_error": row.last_error,
            "http_status": row.http_status,
            "created_at": row.created_at.isoformat(),
            "delivered_at": row.delivered_at.isoformat() if row.delivered_at else None,
        }


async def get_delivery_stats(*, tenant_id: str | None = None) -> dict[str, Any]:
    """Get aggregate webhook delivery statistics, optionally scoped to a tenant."""
    from sqlalchemy import func as sqlfunc

    async with get_session_factory()() as session:
        stmt = select(
            WebhookEventRow.status,
            sqlfunc.count(WebhookEventRow.id).label("count"),
        ).group_by(WebhookEventRow.status)
        if tenant_id is not None:
            stmt = stmt.where(WebhookEventRow.tenant_id == tenant_id)
        result = await session.execute(stmt)
        counts = {row.status: row.count for row in result}

    return {
        "pending": counts.get("pending", 0),
        "delivered": counts.get("delivered", 0),
        "dead_letter": counts.get("dead_letter", 0),
        "total": sum(counts.values()),
    }


async def list_events(
    status: str | None = None,
    event_type: str | None = None,
    tenant_id: str | None = None,
    limit: int = 50,
    offset: int = 0,
) -> tuple[list[dict[str, Any]], int]:
    """List webhook events with filtering and pagination.

    Returns (events, total_count) for operator visibility.
    """
    from sqlalchemy import func as sqlfunc

    async with get_session_factory()() as session:
        # Base query
        query = select(WebhookEventRow).order_by(WebhookEventRow.created_at.desc())
        count_query = select(sqlfunc.count(WebhookEventRow.id))

        # Apply filters
        if status:
            query = query.where(WebhookEventRow.status == status)
            count_query = count_query.where(WebhookEventRow.status == status)
        if event_type:
            query = query.where(WebhookEventRow.event == event_type)
            count_query = count_query.where(WebhookEventRow.event == event_type)
        if tenant_id:
            query = query.where(WebhookEventRow.tenant_id == tenant_id)
            count_query = count_query.where(WebhookEventRow.tenant_id == tenant_id)

        # Get total count
        total = await session.scalar(count_query) or 0

        # Apply pagination
        query = query.limit(limit).offset(offset)
        result = await session.execute(query)
        rows = result.scalars().all()

        events = [
            {
                "id": str(row.id),
                "event": row.event,
                "status": row.status,
                "attempts": row.attempts,
                "max_attempts": row.max_attempts,
                "last_attempt_at": row.last_attempt_at.isoformat() if row.last_attempt_at else None,
                "next_attempt_at": row.next_attempt_at.isoformat() if row.next_attempt_at else None,
                "last_error": row.last_error,
                "http_status": row.http_status,
                "created_at": row.created_at.isoformat(),
                "delivered_at": row.delivered_at.isoformat() if row.delivered_at else None,
                "tenant_id": row.tenant_id,
            }
            for row in rows
        ]

        return events, total


# Statuses safe to delete on operator request — `pending` events may still
# be in flight in the delivery loop, so they're excluded.
TERMINAL_WEBHOOK_STATUSES = ("delivered", "dead_letter")


async def purge_events(
    status: str | None = None,
    event_type: str | None = None,
    tenant_id: str | None = None,
) -> int:
    """Operator-triggered delete for terminal webhook events.

    At least one filter must be supplied (otherwise this would wipe every
    delivered + dead-letter event in one call). Status, when given, must
    be a terminal state. Returns the number of rows deleted.
    """
    if not (status or event_type or tenant_id):
        raise ValueError("at least one filter is required")
    if status and status not in TERMINAL_WEBHOOK_STATUSES:
        raise ValueError(
            f"status must be one of {TERMINAL_WEBHOOK_STATUSES}; got {status!r}"
        )

    async with get_session_factory()() as session:
        stmt = delete(WebhookEventRow).where(
            WebhookEventRow.status.in_(
                [status] if status else list(TERMINAL_WEBHOOK_STATUSES)
            )
        )
        if event_type:
            stmt = stmt.where(WebhookEventRow.event == event_type)
        if tenant_id:
            stmt = stmt.where(WebhookEventRow.tenant_id == tenant_id)
        result = await session.execute(stmt)
        await session.commit()
        count = result.rowcount or 0  # type: ignore[attr-defined]
        logger.info(
            "webhook_events_purged",
            deleted=count,
            status=status,
            event_type=event_type,
            tenant_id=tenant_id,
        )
        return count
