import uuid

from confluent_kafka import Producer
import json
from django.conf import settings
from django.db import transaction
from django.utils import timezone

from api.models import KafkaLog

host = settings.KAFKA_CONFIG["HOST"]
port = str(settings.KAFKA_CONFIG["PORT"])
producer = None


def _producer():
    global producer
    if producer is None:
        producer = Producer({"bootstrap.servers": host + ":" + port})
    return producer


def send_request(topic, data, socket_id, method, *, _durable=True, request_id=None):
    """
    If need use chunk function, make sure the key inside data only has:
    session_id + one list key
    """
    if _durable and transaction.get_connection().in_atomic_block:
        from api.services.integration.delivery import enqueue_kafka_delivery

        enqueue_kafka_delivery(topic=topic, data=data, socket_id=socket_id, method=method)
        return

    MAX_BYTES = 870400

    list_key = None
    for key, value in data.items():
        if key != "session_id" and isinstance(value, list):
            list_key = key
            break

    full_payload_bytes = len(json.dumps(data).encode("utf-8"))

    # Small payload or no list: send immediately
    if full_payload_bytes <= MAX_BYTES or not list_key or not data[list_key]:
        errors = []
        _execute_produce(topic, data, socket_id, method, flush=False, request_id=request_id, errors=errors)
        remaining = _producer().flush()
        if remaining or errors:
            raise RuntimeError(errors[0] if errors else f"Kafka flush left {remaining} message(s)")
        return

    original_list = data[list_key]
    total_items = len(original_list)

    # 1. Calculate average size per item mathematically
    avg_bytes_per_item = full_payload_bytes / total_items

    # 2. Determine a highly conservative, safe chunk size limit
    # We add a 15% safety buffer to account for data skewness
    chunk_size = int((MAX_BYTES / avg_bytes_per_item) * 0.85)
    if chunk_size < 1:
        chunk_size = 1  # Guard against massive individual entries

    print(
        f"Payload size ({full_payload_bytes} bytes) exceeds limit. "
        f"Fast slicing list key '{list_key}' into blocks of {chunk_size}..."
    )

    # 3. Slice through array natively in memory (Runs in milliseconds!)
    errors = []
    for chunk_index, i in enumerate(range(0, total_items, chunk_size)):
        chunk = original_list[i:i + chunk_size]

        safe_data = data.copy()
        safe_data[list_key] = chunk

        chunk_request_id = f"{request_id}:{chunk_index}" if request_id else None
        _execute_produce(
            topic,
            safe_data,
            socket_id,
            method,
            flush=False,
            request_id=chunk_request_id,
            errors=errors,
        )

    # Flush the producer pipeline once all chunks are staged
    remaining = _producer().flush()
    if remaining or errors:
        raise RuntimeError(errors[0] if errors else f"Kafka flush left {remaining} message(s)")

def _execute_produce(topic, data, socket_id, method, flush=True, request_id=None, errors=None):
    """
    Internal helper function containing execution logic
    to send a single safe message packet to Kafka.
    """
    produce_data = data.copy()

    request_id = request_id or str(uuid.uuid4())
    produce_data["request_id"] = request_id

    headers = [
        ("method", method.encode("utf-8")),
    ]

    def delivered(error, message):
        if error is not None and errors is not None:
            errors.append(str(error))

    _producer().produce(
        topic,
        json.dumps(produce_data).encode("utf-8"),
        headers=headers,
        on_delivery=delivered,
    )

    KafkaLog.objects.get_or_create(
        request_id=request_id,
        defaults={
            "request_topic": topic,
            "request_data": json.dumps(produce_data),
            "socket_id": socket_id,
            "request_header": str(headers),
            "request_at": timezone.now(),
        },
    )

    if flush:
        remaining = _producer().flush()
        if remaining or (errors and len(errors)):
            raise RuntimeError(errors[0] if errors else f"Kafka flush left {remaining} message(s)")