import logging
import random
from datetime import timedelta

from django.conf import settings
from django.db import transaction
from django.db.models import Q
from django.utils import timezone

from api.models import PostCommitDelivery
from api.services.integration.context import current_context


logger = logging.getLogger("api.integration.delivery")


def enqueue_kafka_delivery(*, topic, data, socket_id, method) -> PostCommitDelivery:
    if not transaction.get_connection().in_atomic_block:
        raise RuntimeError("Durable post-commit delivery must be enqueued inside transaction.atomic()")
    context = current_context()
    delivery = PostCommitDelivery.objects.create(
        channel="kafka",
        target=topic,
        method=method,
        socket_id=socket_id,
        payload=data,
        correlation_id=context.correlation_id,
    )
    transaction.on_commit(lambda: deliver_by_id(delivery.pk), robust=True)
    return delivery


def _failed(delivery, error):
    config = settings.RESOURCE_BOOKING_INTEGRATION
    now = timezone.now()
    message = f"{error.__class__.__name__}: {error}"[:4000]
    if delivery.attempts >= config["MAX_ATTEMPTS"]:
        PostCommitDelivery.objects.filter(pk=delivery.pk).update(
            status=PostCommitDelivery.Status.DEAD_LETTER,
            claimed_at=None,
            next_attempt_at=None,
            dead_letter_at=now,
            last_error=message,
        )
        return
    delay = min(
        config["RETRY_MAX_SECONDS"],
        config["RETRY_BASE_SECONDS"] * (2 ** max(0, delivery.attempts - 1)),
    )
    jitter = random.Random(str(delivery.delivery_id)).uniform(0, max(1, delay * 0.2))
    PostCommitDelivery.objects.filter(pk=delivery.pk).update(
        status=PostCommitDelivery.Status.RETRY,
        claimed_at=None,
        next_attempt_at=now + timedelta(seconds=delay + jitter),
        last_error=message,
    )


def deliver_by_id(delivery_id) -> bool:
    with transaction.atomic():
        delivery = PostCommitDelivery.objects.select_for_update().filter(pk=delivery_id).first()
        if not delivery or delivery.status not in {
            PostCommitDelivery.Status.PENDING,
            PostCommitDelivery.Status.RETRY,
        }:
            return False
        if delivery.next_attempt_at and delivery.next_attempt_at > timezone.now():
            return False
        delivery.status = PostCommitDelivery.Status.PUBLISHING
        delivery.claimed_at = timezone.now()
        delivery.attempts += 1
        delivery.save(update_fields=["status", "claimed_at", "attempts"])
    try:
        if delivery.channel != "kafka":
            raise RuntimeError(f"Unsupported durable delivery channel {delivery.channel!r}")
        from backend.kafka import send_request

        send_request(
            delivery.target,
            delivery.payload,
            delivery.socket_id,
            delivery.method,
            _durable=False,
            request_id=str(delivery.delivery_id),
        )
    except Exception as error:
        _failed(delivery, error)
        logger.exception(
            "post_commit_delivery_failed",
            extra={"delivery_id": str(delivery.delivery_id), "attempt": delivery.attempts},
        )
        return False
    PostCommitDelivery.objects.filter(pk=delivery.pk).update(
        status=PostCommitDelivery.Status.PUBLISHED,
        claimed_at=None,
        next_attempt_at=None,
        published_at=timezone.now(),
        last_error=None,
    )
    return True


def deliver_pending(limit=100) -> int:
    now = timezone.now()
    claim_timeout = settings.RESOURCE_BOOKING_INTEGRATION["CLAIM_TIMEOUT_SECONDS"]
    PostCommitDelivery.objects.filter(
        status=PostCommitDelivery.Status.PUBLISHING,
        claimed_at__lt=now - timedelta(seconds=claim_timeout),
    ).update(
        status=PostCommitDelivery.Status.RETRY,
        claimed_at=None,
        next_attempt_at=now,
        last_error="Recovered abandoned post-commit delivery claim",
    )
    ids = list(
        PostCommitDelivery.objects.filter(
            status__in=[PostCommitDelivery.Status.PENDING, PostCommitDelivery.Status.RETRY]
        )
        .filter(Q(next_attempt_at__isnull=True) | Q(next_attempt_at__lte=now))
        .order_by("created_at")
        .values_list("id", flat=True)[:limit]
    )
    return sum(int(deliver_by_id(delivery_id)) for delivery_id in ids)