"""Shared plumbing for single-task minions.

Each minion in this project follows the same shape:

    async def handle(event: dict) -> None: ...

    if __name__ == "__main__":
        asyncio.run(run_minion(
            topic="...",
            group_id="...",
            handler=handle,
            dlq_topic="...",
            max_retries=5,
        ))

`run_minion` owns the Kafka consumer boilerplate, signal handling, retry
counter bookkeeping and DLQ routing so the business code in each minion
stays pure: consume one topic, do one job, produce one result topic.
"""
from __future__ import annotations

import asyncio
import json
import signal
from typing import Any, Awaitable, Callable

from aiokafka import AIOKafkaConsumer

from app.config import get_settings
from app.kafka_client import get_producer
from app.logger import get_logger

Handler = Callable[[dict], Awaitable[None]]


async def emit(topic: str, key: str, value: dict) -> None:
    producer = await get_producer()
    await producer.publish(topic=topic, key=key, value=value)


async def send_to_dlq(dlq_topic: str, event: dict, error: str) -> None:
    """Poison-message safety valve: after N failed retries an event lands
    on its step's `.dlq` topic with the last error attached, so a human can
    inspect / replay it."""
    log = get_logger("dlq")
    payload = {
        "event": event,
        "error": error,
        "attempts": event.get("attempts", 0),
    }
    key = f"{event.get('job_id', '?')}:{event.get('licence_number', '?')}"
    log.error("→ DLQ topic=%s key=%s error=%s", dlq_topic, key, error)
    await emit(dlq_topic, key, payload)


async def run_minion(
    *,
    name: str,
    topic: str,
    group_id: str,
    handler: Handler,
    dlq_topic: str | None = None,
    max_retries: int = 5,
) -> None:
    s = get_settings()
    log = get_logger(name)
    consumer = AIOKafkaConsumer(
        topic,
        bootstrap_servers=s.kafka_bootstrap_servers,
        group_id=group_id,
        client_id=f"{s.kafka_client_id}-{name}",
        enable_auto_commit=False,
        auto_offset_reset="earliest",
        value_deserializer=lambda v: json.loads(v.decode("utf-8")),
    )
    await consumer.start()
    log.info("Minion started. topic=%s group=%s", topic, group_id)

    stop = asyncio.Event()

    def _sig(*_):  # noqa: ANN001
        stop.set()

    for sig in (signal.SIGINT, signal.SIGTERM):
        try:
            asyncio.get_event_loop().add_signal_handler(sig, _sig)
        except NotImplementedError:
            pass

    try:
        while not stop.is_set():
            batch = await consumer.getmany(timeout_ms=1000, max_records=20)
            for _tp, msgs in batch.items():
                for msg in msgs:
                    event = msg.value
                    try:
                        await handler(event)
                    except Exception as exc:  # noqa: BLE001
                        event["attempts"] = int(event.get("attempts", 0)) + 1
                        log.exception(
                            "handler failed (attempt %s) key=%s: %s",
                            event["attempts"], msg.key, exc,
                        )
                        if dlq_topic and event["attempts"] >= max_retries:
                            await send_to_dlq(dlq_topic, event, str(exc))
                        # If no DLQ configured, the exception has been logged;
                        # we still commit so the pipeline keeps moving.
                await consumer.commit()
    finally:
        await consumer.stop()
        log.info("Minion stopped.")


def envelope_key(event: dict[str, Any]) -> str:
    """Deterministic Kafka key so all events for the same tank land on the
    same partition — preserves order across steps."""
    return f"{event.get('job_id', '?')}:{event.get('licence_number', '?')}"