import asyncio
import json
import logging

import sqlalchemy as sa

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

logger = logging.getLogger("audit")

GROUP = "audit"
CONSUMER = "audit-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):
    async with Session() as session:
        await session.execute(
            sa.text(
                "INSERT INTO audit_events (transaction_id, event_type, payload, recorded_at) "
                "VALUES (:tx_id, 'transaction_processed', CAST(:payload AS jsonb), now())"
            ),
            {"tx_id": fields.get("transaction_id"), "payload": json.dumps(fields)},
        )
        await session.commit()


async def run():
    """Second, independent consumer group on the same stream as `notifier` --
    proves the async fork fans out to genuinely separate consumers, not one
    consumer doing both jobs."""
    await ensure_group()
    logger.info("audit consumer started")
    while True:
        try:
            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("audit consumer read failed")
            await asyncio.sleep(1)
            continue

        if not resp:
            continue

        for _, messages in resp:
            for msg_id, fields in messages:
                try:
                    await _handle(fields)
                except Exception:
                    logger.exception(f"audit failed to process message {msg_id}")
                await redis_client.xack(settings.stream_name, GROUP, msg_id)