import asyncio
import random
import uuid

import sqlalchemy as sa
import structlog
from opentelemetry import propagate, trace
from opentelemetry.instrumentation.utils import suppress_instrumentation
from opentelemetry.trace import SpanKind, StatusCode

from common.db import Session
from common.redis_client import redis_client
from common.settings import settings

from . import chaos

logger = structlog.get_logger("notifier")
tracer = trace.get_tracer(__name__)

GROUP = "notifier"
CONSUMER = "notifier-1"


async def ensure_group():
    try:
        await redis_client.xgroup_create(settings.stream_name, GROUP, id="$", mkstream=True)
    except Exception as exc:
        if "BUSYGROUP" not in str(exc):
            raise


async def _handle(fields: dict):
    if await chaos.slow_consumer_enabled():
        await asyncio.sleep(chaos.SLEEP_SECONDS)

    async with Session() as session:
        await session.execute(
            sa.text(
                "INSERT INTO notifications (id, transaction_id, channel, status, attempts, sent_at) "
                "VALUES (:id, :tx_id, :channel, 'sent', 1, now())"
            ),
            {
                "id": str(uuid.uuid4()),
                "tx_id": fields.get("transaction_id"),
                "channel": random.choice(["email", "push", "sms"]),
            },
        )
        await session.commit()


async def run():
    """Background consumer loop for the `notifier` consumer group -- a second,
    independent group (`audit`) reads the same stream in services/audit/worker.py.
    Redis Streams carry no automatic context propagation, deliberately."""
    await ensure_group()
    logger.info("consumer_started", consumer_group=GROUP)
    while True:
        try:
            # The blocking poll would otherwise be a ~2s root trace every loop.
            with suppress_instrumentation():
                resp = await redis_client.xreadgroup(
                    groupname=GROUP,
                    consumername=CONSUMER,
                    streams={settings.stream_name: ">"},
                    count=10,
                    block=2000,
                )
        except asyncio.CancelledError:
            raise
        except Exception:
            logger.exception("stream_read_failed", consumer_group=GROUP)
            await asyncio.sleep(1)
            continue

        if not resp:
            continue

        for _, messages in resp:
            for msg_id, fields in messages:
                # Parent is the gateway's PRODUCER span, carried in the message.
                with tracer.start_as_current_span(
                    f"process {settings.stream_name}",
                    context=propagate.extract(fields),
                    kind=SpanKind.CONSUMER,
                    attributes={
                        "messaging.system": "redis",
                        "messaging.destination.name": settings.stream_name,
                        "messaging.operation.type": "process",
                        "messaging.consumer.group.name": GROUP,
                        "messaging.message.id": msg_id,
                    },
                ) as span:
                    try:
                        await _handle(fields)
                        logger.info(
                            "stream_message_processed",
                            consumer_group=GROUP,
                            message_id=msg_id,
                            transaction_id=fields.get("transaction_id"),
                        )
                    except Exception as exc:
                        span.record_exception(exc)
                        span.set_status(StatusCode.ERROR, str(exc))
                        logger.exception(
                            "stream_message_failed",
                            consumer_group=GROUP,
                            message_id=msg_id,
                            transaction_id=fields.get("transaction_id"),
                        )
                    await redis_client.xack(settings.stream_name, GROUP, msg_id)