from __future__ import annotations

import json
import os
import time
from contextlib import contextmanager
from datetime import datetime, timezone
from typing import Any, Iterator

from sqlalchemy import select
from sqlalchemy.orm import Session

from services.shared.core import new_id, utc_now_iso
from services.shared.models import EventEnvelope, EventOutboxItem
from services.shared.sql_models import EventInboxRow, EventOutboxRow

EXCHANGE_NAME = "mvpcc.domain.events"
AUDIT_QUEUE = "mvpcc.audit.events"
REPORTING_QUEUE = "mvpcc.reporting.events"
PREFETCH_COUNT = 20
RETRY_SCHEDULE = (5, 15, 60, 300)


def _bool_env(name: str, default: bool) -> bool:
    raw = os.getenv(name)
    if raw is None:
        return default
    return raw.strip().lower() in {"1", "true", "yes", "on"}


def event_bus_enabled() -> bool:
    return _bool_env("EVENT_BUS_ENABLED", False)


def consumer_enabled() -> bool:
    return _bool_env("EVENT_BUS_CONSUMER_ENABLED", True)


def event_bus_url() -> str:
    return os.getenv("EVENT_BUS_URL", "amqp://guest:guest@localhost:5672/").strip()


def event_bus_exchange() -> str:
    return os.getenv("EVENT_BUS_EXCHANGE", EXCHANGE_NAME).strip() or EXCHANGE_NAME


def event_bus_audit_queue() -> str:
    return os.getenv("EVENT_BUS_AUDIT_QUEUE", AUDIT_QUEUE).strip() or AUDIT_QUEUE


def event_bus_reporting_queue() -> str:
    return os.getenv("EVENT_BUS_REPORTING_QUEUE", REPORTING_QUEUE).strip() or REPORTING_QUEUE


def event_bus_dispatch_batch_size() -> int:
    try:
        return max(1, int(os.getenv("EVENT_BUS_DISPATCH_BATCH_SIZE", "50")))
    except ValueError:
        return 50


def event_bus_max_attempts() -> int:
    try:
        return max(1, int(os.getenv("EVENT_BUS_MAX_ATTEMPTS", "5")))
    except ValueError:
        return 5


def event_bus_poll_interval_seconds() -> float:
    try:
        return max(0.25, float(os.getenv("EVENT_BUS_POLL_INTERVAL_SECONDS", "2")))
    except ValueError:
        return 2.0


def _lazy_import_pika():
    try:
        import pika  # type: ignore
    except ImportError as exc:
        raise RuntimeError("pika is required for EVENT_BUS_ENABLED=1") from exc
    return pika


def _iso_in(seconds: int) -> str:
    dt = datetime.now(timezone.utc).timestamp() + seconds
    return datetime.fromtimestamp(dt, tz=timezone.utc).replace(microsecond=0).isoformat()


def retry_delay_seconds(attempt_count: int) -> int:
    if attempt_count <= 1:
        return RETRY_SCHEDULE[0]
    if attempt_count == 2:
        return RETRY_SCHEDULE[1]
    if attempt_count == 3:
        return RETRY_SCHEDULE[2]
    return RETRY_SCHEDULE[3]


def routing_key_for(event_type: str) -> str:
    return event_type.strip()


def build_envelope(
    *,
    event_type: str,
    producer: str,
    entity_type: str,
    entity_id: str,
    payload: dict[str, Any],
    correlation_id: str | None = None,
    event_id: str | None = None,
    event_version: int = 1,
) -> EventEnvelope:
    return EventEnvelope(
        event_id=event_id or new_id("evt"),
        event_type=event_type,
        event_version=event_version,
        occurred_at=utc_now_iso(),
        producer=producer,
        entity_type=entity_type,
        entity_id=entity_id,
        correlation_id=correlation_id,
        routing_key=routing_key_for(event_type),
        payload=payload,
    )


def outbox_item_from_row(row: EventOutboxRow) -> EventOutboxItem:
    envelope = json.loads(row.payload_json or "{}")
    return EventOutboxItem(
        event_id=row.event_id,
        event_type=row.event_type,
        event_version=row.event_version,
        producer_service=row.producer_service,
        entity_type=row.entity_type,
        entity_id=row.entity_id,
        correlation_id=row.correlation_id,
        routing_key=row.routing_key,
        payload=envelope.get("payload") if isinstance(envelope.get("payload"), dict) else {},
        status=row.status,
        attempt_count=row.attempt_count,
        last_error=row.last_error,
        available_at=row.available_at,
        published_at=row.published_at,
        created_at=row.created_at,
        updated_at=row.updated_at,
    )


def append_outbox_event(
    session: Session,
    *,
    event_type: str,
    producer_service: str,
    entity_type: str,
    entity_id: str,
    payload: dict[str, Any],
    correlation_id: str | None = None,
) -> EventOutboxRow:
    envelope = build_envelope(
        event_type=event_type,
        producer=producer_service,
        entity_type=entity_type,
        entity_id=entity_id,
        payload=payload,
        correlation_id=correlation_id,
    )
    now = utc_now_iso()
    row = EventOutboxRow(
        event_id=envelope.event_id,
        event_type=envelope.event_type,
        event_version=envelope.event_version,
        producer_service=envelope.producer,
        entity_type=envelope.entity_type,
        entity_id=envelope.entity_id,
        correlation_id=envelope.correlation_id,
        routing_key=envelope.routing_key,
        payload_json=json.dumps(envelope.model_dump(), ensure_ascii=False),
        status="pending",
        attempt_count=0,
        last_error=None,
        available_at=now,
        published_at=None,
        created_at=now,
        updated_at=now,
    )
    session.add(row)
    return row


def list_publishable_outbox(session: Session, limit: int | None = None) -> list[EventOutboxRow]:
    now = utc_now_iso()
    stmt = (
        select(EventOutboxRow)
        .where(
            EventOutboxRow.available_at <= now,
            EventOutboxRow.status.in_(["pending", "failed"]),
        )
        .order_by(EventOutboxRow.id.asc())
    )
    rows = session.execute(stmt).scalars().all()
    if limit is None:
        limit = event_bus_dispatch_batch_size()
    return rows[:limit]


def mark_outbox_published(session: Session, row: EventOutboxRow) -> None:
    now = utc_now_iso()
    row.status = "published"
    row.published_at = now
    row.updated_at = now
    row.last_error = None


def mark_outbox_failed(session: Session, row: EventOutboxRow, error: str) -> None:
    row.attempt_count += 1
    row.status = "failed"
    row.last_error = error[:2000]
    row.available_at = _iso_in(retry_delay_seconds(row.attempt_count))
    row.updated_at = utc_now_iso()


def retry_outbox_event(session: Session, row: EventOutboxRow) -> None:
    now = utc_now_iso()
    row.status = "pending"
    row.last_error = None
    row.available_at = now
    row.updated_at = now


def inbox_seen(session: Session, consumer_name: str, event_id: str) -> bool:
    row = session.execute(
        select(EventInboxRow).where(
            EventInboxRow.consumer_name == consumer_name,
            EventInboxRow.event_id == event_id,
        )
    ).scalar_one_or_none()
    return row is not None


def record_inbox(
    session: Session,
    *,
    consumer_name: str,
    event_id: str,
    event_type: str,
    status: str,
    notes: str | None = None,
) -> EventInboxRow:
    row = EventInboxRow(
        consumer_name=consumer_name,
        event_id=event_id,
        event_type=event_type,
        processed_at=utc_now_iso(),
        status=status,
        notes=notes,
    )
    session.add(row)
    return row


@contextmanager
def rabbitmq_connection() -> Iterator[tuple[Any, Any]]:
    pika = _lazy_import_pika()
    params = pika.URLParameters(event_bus_url())
    connection = pika.BlockingConnection(params)
    channel = connection.channel()
    try:
        yield connection, channel
    finally:
        try:
            connection.close()
        except Exception:
            pass


def ensure_rabbitmq_topology(channel: Any) -> None:
    exchange = event_bus_exchange()
    audit_queue = event_bus_audit_queue()
    reporting_queue = event_bus_reporting_queue()
    channel.exchange_declare(exchange=exchange, exchange_type="topic", durable=True)
    for queue_name, bindings in (
        (audit_queue, ["interaction.*", "agent.state.changed", "call.recording.ready", "ivr.completed"]),
        (
            reporting_queue,
            ["interaction.created", "interaction.closed", "agent.state.changed", "ivr.completed", "call.recording.ready"],
        ),
    ):
        dlq_name = f"{queue_name}.dlq"
        channel.queue_declare(queue=dlq_name, durable=True)
        channel.queue_declare(
            queue=queue_name,
            durable=True,
            arguments={
                "x-dead-letter-exchange": "",
                "x-dead-letter-routing-key": dlq_name,
            },
        )
        for binding in bindings:
            channel.queue_bind(queue=queue_name, exchange=exchange, routing_key=binding)


def publish_envelope(channel: Any, envelope: dict[str, Any], headers: dict[str, Any] | None = None) -> None:
    pika = _lazy_import_pika()
    body = json.dumps(envelope, ensure_ascii=False).encode("utf-8")
    properties = pika.BasicProperties(
        content_type="application/json",
        delivery_mode=2,
        headers=headers or {},
    )
    channel.basic_publish(
        exchange=event_bus_exchange(),
        routing_key=envelope["routing_key"],
        body=body,
        properties=properties,
    )


def dispatch_outbox_batch(session: Session, limit: int | None = None) -> dict[str, int]:
    rows = list_publishable_outbox(session, limit=limit)
    if not rows:
        return {"claimed": 0, "published": 0, "failed": 0}

    published = 0
    failed = 0
    try:
        with rabbitmq_connection() as (_, channel):
            ensure_rabbitmq_topology(channel)
            for row in rows:
                try:
                    publish_envelope(channel, json.loads(row.payload_json or "{}"))
                    mark_outbox_published(session, row)
                    published += 1
                except Exception as exc:  # noqa: BLE001
                    mark_outbox_failed(session, row, str(exc))
                    failed += 1
            session.commit()
    except Exception as exc:  # noqa: BLE001
        for row in rows:
            mark_outbox_failed(session, row, str(exc))
        session.commit()
        failed = len(rows)
        published = 0
    return {"claimed": len(rows), "published": published, "failed": failed}


def consume_one_message(
    queue_name: str,
    handler,
    *,
    max_attempts: int | None = None,
) -> bool:
    if max_attempts is None:
        max_attempts = event_bus_max_attempts()
    with rabbitmq_connection() as (_, channel):
        ensure_rabbitmq_topology(channel)
        channel.basic_qos(prefetch_count=PREFETCH_COUNT)
        method, properties, body = channel.basic_get(queue=queue_name, auto_ack=False)
        if not method:
            return False

        headers = dict((properties.headers or {}))
        try:
            envelope = json.loads(body.decode("utf-8"))
            handler(envelope)
            channel.basic_ack(delivery_tag=method.delivery_tag)
            return True
        except Exception as exc:  # noqa: BLE001
            retry_count = int(headers.get("x-retry-count", 0))
            if retry_count + 1 < max_attempts:
                next_headers = dict(headers)
                next_headers["x-retry-count"] = retry_count + 1
                publish_envelope(channel, json.loads(body.decode("utf-8")), headers=next_headers)
                channel.basic_ack(delivery_tag=method.delivery_tag)
            else:
                dlq_name = f"{queue_name}.dlq"
                pika = _lazy_import_pika()
                channel.basic_publish(
                    exchange="",
                    routing_key=dlq_name,
                    body=body,
                    properties=pika.BasicProperties(
                        content_type="application/json",
                        delivery_mode=2,
                        headers={**headers, "error": str(exc)[:2000]},
                    ),
                )
                channel.basic_ack(delivery_tag=method.delivery_tag)
            return False


def poll_forever(loop_fn) -> None:
    while True:
        try:
            loop_fn()
        except Exception:
            time.sleep(event_bus_poll_interval_seconds())
        else:
            time.sleep(event_bus_poll_interval_seconds())