import json
import logging
import random
from dataclasses import dataclass
from datetime import timedelta

import requests
from confluent_kafka import Producer
from django.conf import settings
from django.db import transaction
from django.db.models import F
from django.utils import timezone

from api.models import (
    IntegrationOutbox,
    IntegrationPublisherState,
)
from api.services.integration.outbox import build_transport_envelope, integration_source_scope


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


class PublishError(RuntimeError):
    pass


def _sanitize_error(error: Exception) -> str:
    message = f"{error.__class__.__name__}: {error}"
    config = settings.RESOURCE_BOOKING_INTEGRATION
    for key in (
        "HTTP_AUTH_TOKEN",
        "KAFKA_BOOTSTRAP_SERVERS",
        "KAFKA_SASL_USERNAME",
        "KAFKA_SASL_PASSWORD",
        "SNAPSHOT_SERVICE_TOKEN",
    ):
        secret = config.get(key)
        if secret:
            message = message.replace(str(secret), "[REDACTED]")
    return message[:4000]


@dataclass(frozen=True)
class PublishResult:
    claimed: int = 0
    published: int = 0
    retried: int = 0
    dead_lettered: int = 0


@dataclass(frozen=True)
class TransportChangeSet:
    """One complete source transaction and therefore one transport record."""

    events: tuple[IntegrationOutbox, ...]

    @property
    def source_scope(self):
        return self.events[0].source_scope

    @property
    def source_sequence(self):
        return self.events[0].transport_sequence

    @property
    def change_set_id(self):
        return self.events[0].change_set_id

    @property
    def payload(self) -> dict:
        return build_transport_envelope(self.events)


class OutboxTransport:
    def publish(self, change_set: TransportChangeSet) -> None:
        raise NotImplementedError


class KafkaOutboxTransport(OutboxTransport):
    def __init__(self):
        config = settings.RESOURCE_BOOKING_INTEGRATION
        self.topic = config["KAFKA_TOPIC"]
        producer_config = {
            "bootstrap.servers": config["KAFKA_BOOTSTRAP_SERVERS"],
            "enable.idempotence": True,
            "acks": "all",
            "max.in.flight.requests.per.connection": 1,
            "security.protocol": config["KAFKA_SECURITY_PROTOCOL"],
            "message.max.bytes": config["MAX_EVENT_BYTES"],
        }
        optional = {
            "sasl.mechanism": config.get("KAFKA_SASL_MECHANISM"),
            "sasl.username": config.get("KAFKA_SASL_USERNAME"),
            "sasl.password": config.get("KAFKA_SASL_PASSWORD"),
            "ssl.ca.location": config.get("KAFKA_SSL_CA_LOCATION"),
        }
        producer_config.update({key: value for key, value in optional.items() if value})
        self.producer = Producer(producer_config)

    def publish(self, change_set: TransportChangeSet) -> None:
        if not self.topic:
            raise PublishError("RB_INTEGRATION_KAFKA_TOPIC is not configured")
        delivery_error = []

        def delivered(error, message):
            if error:
                delivery_error.append(str(error))

        self.producer.produce(
            self.topic,
            json.dumps(change_set.payload, sort_keys=True, separators=(",", ":")).encode("utf-8"),
            # One source-scope key forces all live events through one Kafka
            # partition, preserving the database commit sequence end to end.
            key=change_set.source_scope.encode("utf-8"),
            headers={
                "event_id": str(change_set.change_set_id),
                "schema_version": change_set.events[0].schema_version,
                "idempotency_key": str(change_set.change_set_id),
                "correlation_id": change_set.events[0].correlation_id or "",
                "source_scope": change_set.source_scope,
                "source_sequence": str(change_set.source_sequence),
                "source_transaction_id": str(change_set.change_set_id),
                "transaction_count": str(len(change_set.events)),
                "transaction_complete": "true",
            },
            on_delivery=delivered,
        )
        remaining = self.producer.flush(10)
        if remaining or delivery_error:
            raise PublishError(delivery_error[0] if delivery_error else f"Kafka flush left {remaining} message(s)")


class HttpOutboxTransport(OutboxTransport):
    def publish(self, change_set: TransportChangeSet) -> None:
        config = settings.RESOURCE_BOOKING_INTEGRATION
        if not config["HTTP_URL"]:
            raise PublishError("RB_INTEGRATION_HTTP_URL is not configured")
        headers = {
            "Content-Type": "application/json",
            "Idempotency-Key": str(change_set.change_set_id),
            "X-Correlation-ID": change_set.events[0].correlation_id
            or str(change_set.change_set_id),
        }
        if config["HTTP_AUTH_TOKEN"]:
            headers["Authorization"] = f"Bearer {config['HTTP_AUTH_TOKEN']}"
        response = requests.post(
            config["HTTP_URL"],
            json=change_set.payload,
            headers=headers,
            timeout=config["HTTP_TIMEOUT_SECONDS"],
        )
        if not 200 <= response.status_code < 300:
            raise PublishError(f"Adapter returned HTTP {response.status_code}")


def configured_transport() -> OutboxTransport:
    transport = settings.RESOURCE_BOOKING_INTEGRATION["TRANSPORT"].lower()
    if transport == "kafka":
        return KafkaOutboxTransport()
    if transport == "http":
        return HttpOutboxTransport()
    raise PublishError(f"Unsupported RB_INTEGRATION_TRANSPORT: {transport!r}")


def recover_abandoned_claims(now=None) -> int:
    now = now or timezone.now()
    threshold = now - timedelta(seconds=settings.RESOURCE_BOOKING_INTEGRATION["CLAIM_TIMEOUT_SECONDS"])
    return IntegrationOutbox.objects.filter(
        status=IntegrationOutbox.Status.PUBLISHING,
        claimed_at__lt=threshold,
    ).update(
        status=IntegrationOutbox.Status.RETRY,
        next_attempt_at=now,
        claimed_at=None,
        last_error="Recovered abandoned publisher claim",
    )


def claim_change_sets(limit=None, now=None) -> list[TransportChangeSet]:
    now = now or timezone.now()
    limit = limit or settings.RESOURCE_BOOKING_INTEGRATION["PUBLISH_BATCH_SIZE"]
    source_scope = integration_source_scope()
    with transaction.atomic():
        publisher_state, _ = IntegrationPublisherState.objects.get_or_create(
            source_scope=source_scope,
            defaults={"status": "claiming"},
        )
        IntegrationPublisherState.objects.select_for_update().get(pk=publisher_state.pk)
        candidate_sequences = list(
            IntegrationOutbox.objects.filter(source_scope=source_scope)
            .exclude(
                status__in=[IntegrationOutbox.Status.PUBLISHED, IntegrationOutbox.Status.SUPERSEDED]
            )
            .order_by("transport_sequence")
            .values_list("transport_sequence", flat=True)
            .distinct()[:limit]
        )
        claimed_change_sets = []
        for source_sequence in candidate_sequences:
            events = list(
                IntegrationOutbox.objects.select_for_update()
                .filter(
                    source_scope=source_scope,
                    transport_sequence=source_sequence,
                )
                .order_by("transaction_index")
            )
            if not events:
                break
            expected_count = events[0].transaction_count
            if (
                not all(event.transaction_finalized for event in events)
                or len(events) != expected_count
                or [event.transaction_index for event in events]
                != list(range(1, expected_count + 1))
                or len({event.change_set_id for event in events}) != 1
            ):
                break
            if any(
                event.status
                not in (IntegrationOutbox.Status.PENDING, IntegrationOutbox.Status.RETRY)
                or (event.next_attempt_at and event.next_attempt_at > now)
                for event in events
            ):
                break
            for event in events:
                event.status = IntegrationOutbox.Status.PUBLISHING
                event.claimed_at = now
                event.attempts += 1
            IntegrationOutbox.objects.bulk_update(events, ["status", "claimed_at", "attempts"])
            claimed_change_sets.append(TransportChangeSet(tuple(events)))
        return claimed_change_sets


def _mark_published(change_set: TransportChangeSet) -> None:
    IntegrationOutbox.objects.filter(
        pk__in=[event.pk for event in change_set.events],
        status=IntegrationOutbox.Status.PUBLISHING,
    ).update(
        status=IntegrationOutbox.Status.PUBLISHED,
        published_at=timezone.now(),
        claimed_at=None,
        next_attempt_at=None,
        last_error=None,
    )


def _publisher_state(*, status: str, error: str | None = None, success=False, sequence=None) -> None:
    now = timezone.now()
    defaults = {
        "status": status,
        "heartbeat_at": now,
        "last_error": error,
    }
    if success:
        defaults["last_success_at"] = now
    if sequence is not None:
        defaults["last_published_sequence"] = sequence
    IntegrationPublisherState.objects.update_or_create(
        source_scope=integration_source_scope(), defaults=defaults
    )


def heartbeat_publisher(status="idle") -> None:
    _publisher_state(status=status)


def _release_unattempted(change_sets: list[TransportChangeSet], *, error: Exception) -> None:
    events = [event for change_set in change_sets for event in change_set.events]
    if not events:
        return
    IntegrationOutbox.objects.filter(
        pk__in=[event.pk for event in events],
        status=IntegrationOutbox.Status.PUBLISHING,
    ).update(
        status=IntegrationOutbox.Status.RETRY,
        attempts=F("attempts") - 1,
        claimed_at=None,
        next_attempt_at=timezone.now(),
        last_error=f"Blocked by earlier transport failure: {_sanitize_error(error)}",
    )


def _mark_failed(change_set: TransportChangeSet, error: Exception) -> bool:
    config = settings.RESOURCE_BOOKING_INTEGRATION
    now = timezone.now()
    attempts = max(event.attempts for event in change_set.events)
    event_ids = [event.pk for event in change_set.events]
    dead_letter = attempts >= config["MAX_ATTEMPTS"]
    if dead_letter:
        IntegrationOutbox.objects.filter(pk__in=event_ids).update(
            status=IntegrationOutbox.Status.DEAD_LETTER,
            claimed_at=None,
            next_attempt_at=None,
            dead_letter_at=now,
            last_error=_sanitize_error(error),
        )
        return True
    base = config["RETRY_BASE_SECONDS"]
    cap = config["RETRY_MAX_SECONDS"]
    delay = min(cap, base * (2 ** max(0, attempts - 1)))
    jitter = random.Random(str(change_set.change_set_id)).uniform(0, max(1, delay * 0.2))
    IntegrationOutbox.objects.filter(pk__in=event_ids).update(
        status=IntegrationOutbox.Status.RETRY,
        claimed_at=None,
        next_attempt_at=now + timedelta(seconds=delay + jitter),
        last_error=_sanitize_error(error),
    )
    return False


def publish_once(*, transport=None, limit=None) -> PublishResult:
    if not settings.RESOURCE_BOOKING_INTEGRATION["PUBLISH_ENABLED"]:
        _publisher_state(status="disabled")
        return PublishResult()
    recover_abandoned_claims()
    change_sets = claim_change_sets(limit=limit)
    if not change_sets:
        heartbeat_publisher()
        return PublishResult()
    transport = transport or configured_transport()
    published = retried = dead_lettered = 0
    for index, change_set in enumerate(change_sets):
        try:
            transport.publish(change_set)
        except Exception as error:
            sanitized_error = _sanitize_error(error)
            dead = _mark_failed(change_set, error)
            retried += int(not dead)
            dead_lettered += int(dead)
            logger.error(
                "integration_outbox_publish_failed",
                extra={
                    "change_set_id": str(change_set.change_set_id),
                    "source_sequence": change_set.source_sequence,
                    "attempt": change_set.events[0].attempts,
                    "dead_letter": dead,
                    "error": sanitized_error,
                },
            )
            _release_unattempted(change_sets[index + 1 :], error=error)
            _publisher_state(status="blocked", error=sanitized_error)
            break
        else:
            _mark_published(change_set)
            published += 1
            _publisher_state(
                status="publishing",
                success=True,
                sequence=change_set.source_sequence,
            )
            logger.info(
                "integration_outbox_published",
                extra={
                    "change_set_id": str(change_set.change_set_id),
                    "source_sequence": change_set.source_sequence,
                    "attempt": change_set.events[0].attempts,
                },
            )
    return PublishResult(
        claimed=len(change_sets),
        published=published,
        retried=retried,
        dead_lettered=dead_lettered,
    )