import asyncio
from datetime import datetime, timezone
from typing import Any

from sqlalchemy import delete, select
from sqlalchemy.ext.asyncio import AsyncSession

from config import PUBLIC_LINK, REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD, SUPERNODE
from database import delete_notification, get_servers, get_tariff_by_id, store_key
from database.models import Key, Server, Tariff
from handlers.utils import check_server_key_limit, get_least_loaded_cluster
from logger import logger
from panels.remnawave import RemnawaveAPI
from panels.three_xui import (
    ClientConfig,
    add_client,
    delete_client,
    extend_client_key,
    get_client_traffic,
    get_xui_instance,
    toggle_client,
)


async def create_key_on_cluster(
    cluster_id: str,
    tg_id: int,
    client_id: str,
    email: str,
    expiry_timestamp: int,
    plan: int = None,
    session: AsyncSession = None,
    remnawave_link: str = None,
    hwid_limit: int = None,
    traffic_limit_bytes: int = None, 
):
    try:
        servers = await get_servers(session, include_enabled=True)
        cluster = servers.get(cluster_id)
        server_id_to_store = cluster_id

        if not cluster:
            found_servers = []
            for _key, server_list in servers.items():
                for server_info in server_list:
                    if server_info.get("server_name", "").lower() == cluster_id.lower():
                        found_servers.append(server_info)
            if found_servers:
                cluster = found_servers
                server_id_to_store = found_servers[0].get("server_name")
            else:
                raise ValueError(
                    f"Кластер или сервер с ID/именем {cluster_id} не найден."
                )

        enabled_servers = [s for s in cluster if s.get("enabled", True)]
        if not enabled_servers:
            logger.warning(
                f"[Key Creation] Нет доступных серверов в кластере {cluster_id}"
            )
            return

        if plan is not None and traffic_limit_bytes is None:
            tariff = await get_tariff_by_id(session, plan)
            if not tariff:
                raise ValueError(f"Тариф с id={plan} не найден.")
            traffic_limit_bytes = (
                int(tariff["traffic_limit"]) if tariff["traffic_limit"] else None
            )
            if hwid_limit is None and tariff.get("device_limit") is not None:
                hwid_limit = int(tariff["device_limit"])

        remnawave_servers = [
            s
            for s in enabled_servers
            if s.get("panel_type", "3x-ui").lower() == "remnawave"
            and await check_server_key_limit(s, session)
        ]
        xui_servers = [
            s
            for s in enabled_servers
            if s.get("panel_type", "3x-ui").lower() == "3x-ui"
            and await check_server_key_limit(s, session)
        ]

        if not remnawave_servers and not xui_servers:
            logger.warning(
                f"[Key Creation] Нет серверов с доступным лимитом в кластере {cluster_id}"
            )
            return

        semaphore = asyncio.Semaphore(2)
        remnawave_created = False
        remnawave_key = None
        remnawave_client_id = None

        if remnawave_servers:
            remna = RemnawaveAPI(remnawave_servers[0]["api_url"])
            logged_in = await remna.login(REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD)
            if not logged_in:
                logger.error("Не удалось войти в Remnawave API")
            else:
                expire_at = (
                    datetime.utcfromtimestamp(expiry_timestamp / 1000).isoformat() + "Z"
                )
                inbound_ids = [
                    s.get("inbound_id")
                    for s in remnawave_servers
                    if s.get("inbound_id")
                ]

                if not inbound_ids:
                    logger.warning("Нет inbound_id у серверов Remnawave")
                else:
                    short_uuid = None
                    if remnawave_link and "/" in remnawave_link:
                        short_uuid = remnawave_link.rstrip("/").split("/")[-1]

                    user_data = {
                        "username": email,
                        "trafficLimitStrategy": "NO_RESET",
                        "expireAt": expire_at,
                        "telegramId": tg_id,
                        "activeUserInbounds": inbound_ids,
                    }

                    if traffic_limit_bytes and traffic_limit_bytes > 0:
                        user_data["trafficLimitBytes"] = traffic_limit_bytes

                    if short_uuid:
                        user_data["shortUuid"] = short_uuid
                    if hwid_limit is not None:
                        user_data["hwidDeviceLimit"] = hwid_limit

                    result = await remna.create_user(user_data)
                    if not result:
                        logger.error("Ошибка при создании пользователя в Remnawave")
                    else:
                        remnawave_created = True
                        remnawave_key = result.get("subscriptionUrl")
                        remnawave_client_id = result.get("uuid")
                        logger.info(
                            f"[Key Creation] Пользователь создан в Remnawave: {result}"
                        )

        public_link = f"{PUBLIC_LINK}{email}/{tg_id}" if xui_servers else None
        final_client_id = remnawave_client_id or client_id
        logger.info(
            f"[Debug] 3x-ui servers для кластера {cluster_id}: {[s['server_name'] for s in xui_servers]}"
        )

        if xui_servers:
            if SUPERNODE:
                for server_info in xui_servers:
                    await create_client_on_server(
                        server_info,
                        tg_id,
                        final_client_id,
                        email,
                        expiry_timestamp,
                        semaphore,
                        plan=plan,
                        session=session,
                    )
            else:
                await asyncio.gather(
                    *[
                        create_client_on_server(
                            server,
                            tg_id,
                            final_client_id,
                            email,
                            expiry_timestamp,
                            semaphore,
                            plan=plan,
                            session=session,
                        )
                        for server in xui_servers
                    ],
                    return_exceptions=True,
                )

        if (remnawave_created and remnawave_client_id) or xui_servers:
            await store_key(
                session=session,
                tg_id=tg_id,
                client_id=final_client_id,
                email=email,
                expiry_time=expiry_timestamp,
                key=public_link,
                server_id=server_id_to_store,
                remnawave_link=remnawave_key,
                tariff_id=plan,
            )

    except Exception as e:
        logger.error(f"Ошибка при создании ключа: {e}")
        raise e


async def create_client_on_server(
    server_info: dict,
    tg_id: int,
    client_id: str,
    email: str,
    expiry_timestamp: int,
    semaphore: asyncio.Semaphore,
    plan: int = None,
    session=None,
):
    """
    Создает клиента на указанном 3x-ui сервере с лимитом по тарифу (через ORM).
    """
    logger.info(
        f"[Client] Вход в create_client_on_server: сервер={server_info.get('server_name')}, план={plan}"
    )

    async with semaphore:
        xui = await get_xui_instance(server_info["api_url"])
        inbound_id = server_info.get("inbound_id")
        server_name = server_info.get("server_name", "unknown")

        if not inbound_id:
            logger.warning(
                f"INBOUND_ID отсутствует для сервера {server_name}. Пропуск."
            )
            return

        if SUPERNODE:
            unique_email = f"{email}_{server_name.lower()}"
            sub_id = email
        else:
            unique_email = email
            sub_id = unique_email

        total_gb_value = 0
        device_limit_value = None

        if plan is not None:
            tariff = await get_tariff_by_id(session, plan)
            logger.info(f"[Tariff Debug] Получен тариф: {tariff}")
            if not tariff:
                raise ValueError(f"Тариф с id={plan} не найден.")

            total_gb_value = (
                int(tariff["traffic_limit"]) if tariff.get("traffic_limit") else 0
            )
            device_limit_value = (
                int(tariff["device_limit"])
                if tariff.get("device_limit") is not None
                else None
            )

        try:
            logger.info(
                f"[Client] Вызов add_client: email={unique_email}, client_id={client_id}, GB={total_gb_value}, Devices={device_limit_value}"
            )

            await add_client(
                xui,
                ClientConfig(
                    client_id=client_id,
                    email=unique_email,
                    tg_id=tg_id,
                    limit_ip=device_limit_value,
                    total_gb=total_gb_value,
                    expiry_time=expiry_timestamp,
                    enable=True,
                    flow="xtls-rprx-vision",
                    inbound_id=int(inbound_id),
                    sub_id=sub_id,
                ),
            )
            logger.info(f"[Client] Клиент успешно добавлен на сервер {server_name}")
        except Exception as e:
            logger.error(
                f"[Client Error] Не удалось создать клиента на {server_name}: {e}"
            )

        if SUPERNODE:
            await asyncio.sleep(0.7)


async def renew_key_in_cluster(
    cluster_id: str,
    email: str,
    client_id: str,
    new_expiry_time: int,
    total_gb: int,
    session: AsyncSession,
    hwid_device_limit: int = None,
):
    try:
        servers = await get_servers(session)
        cluster = servers.get(cluster_id)

        if not cluster:
            found_servers = []
            for _key, server_list in servers.items():
                for server_info in server_list:
                    if server_info.get("server_name", "").lower() == cluster_id.lower():
                        found_servers.append(server_info)
            if found_servers:
                cluster = found_servers
            else:
                raise ValueError(
                    f"Кластер или сервер с ID/именем {cluster_id} не найден."
                )

        result = await session.execute(
            select(Key.tg_id, Key.server_id).where(Key.client_id == client_id).limit(1)
        )
        row = result.first()
        if not row:
            logger.error(
                f"Не найден пользователь с client_id={client_id} в таблице keys."
            )
            return False

        tg_id, server_id = row

        result = await session.execute(
            select(Server.tariff_group).where(Server.server_name == server_id)
        )
        tariff_group_row = result.scalar_one_or_none()

        if tariff_group_row:
            result = await session.execute(
                select(Tariff)
                .where(Tariff.group_code == tariff_group_row, Tariff.is_active is True)
                .order_by(Tariff.duration_days.desc())
                .limit(1)
            )
            tariff = result.scalar_one_or_none()
            if tariff and tariff.device_limit is not None:
                hwid_device_limit = int(tariff.device_limit)

        notification_prefixes = ["key_24h", "key_10h", "key_expired", "renew"]
        for notif in notification_prefixes:
            notification_id = f"{email}_{notif}"
            await delete_notification(session, tg_id, notification_id)
        logger.info(f"🧹 Уведомления для ключа {email} очищены при продлении.")

        remnawave_inbound_ids = []
        tasks = []
        for server_info in cluster:
            if server_info.get("panel_type", "3x-ui").lower() == "remnawave":
                inbound_id = server_info.get("inbound_id")
                if inbound_id:
                    remnawave_inbound_ids.append(inbound_id)

        if remnawave_inbound_ids:
            remnawave_server = next(
                (
                    s
                    for s in cluster
                    if s.get("panel_type", "").lower() == "remnawave"
                    and s.get("inbound_id") in remnawave_inbound_ids
                ),
                None,
            )
            if remnawave_server:
                remna = RemnawaveAPI(remnawave_server["api_url"])
                if await remna.login(REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD):
                    expire_iso = (
                        datetime.utcfromtimestamp(new_expiry_time // 1000).isoformat()
                        + "Z"
                    )
                    updated = await remna.update_user(
                        uuid=client_id,
                        expire_at=expire_iso,
                        active_user_inbounds=remnawave_inbound_ids,
                        traffic_limit_bytes=total_gb,
                        hwid_device_limit=hwid_device_limit,
                    )
                    if updated:
                        logger.info(f"Подписка Remnawave {client_id} успешно продлена")
                        await remna.reset_user_traffic(client_id)
                    else:
                        logger.warning(
                            f"Не удалось продлить подписку Remnawave {client_id}"
                        )
                else:
                    logger.error("Не удалось войти в Remnawave API")

        for server_info in cluster:
            if server_info.get("panel_type", "3x-ui").lower() != "3x-ui":
                continue

            xui = await get_xui_instance(server_info["api_url"])
            inbound_id = server_info.get("inbound_id")
            server_name = server_info.get("server_name", "unknown")

            if not inbound_id:
                logger.warning(
                    f"INBOUND_ID отсутствует для сервера {server_name}. Пропуск."
                )
                continue

            if SUPERNODE:
                unique_email = f"{email}_{server_name.lower()}"
                sub_id = email
            else:
                unique_email = email
                sub_id = unique_email

            tasks.append(
                extend_client_key(
                    xui=xui,
                    inbound_id=int(inbound_id),
                    email=unique_email,
                    new_expiry_time=new_expiry_time,
                    client_id=client_id,
                    total_gb=total_gb,
                    sub_id=sub_id,
                    tg_id=tg_id,
                    limit_ip=hwid_device_limit,
                )
            )

        await asyncio.gather(*tasks, return_exceptions=True)

    except Exception as e:
        logger.error(
            f"Не удалось продлить ключ {client_id} в кластере/на сервере {cluster_id}: {e}"
        )
        raise


async def delete_key_from_cluster(
    cluster_id: str, email: str, client_id: str, session: AsyncSession
):
    """Удаление ключа с серверов в кластере или с конкретного сервера"""
    try:
        servers = await get_servers(session)
        cluster = servers.get(cluster_id)

        if not cluster:
            found_servers = []
            for _, server_list in servers.items():
                for server_info in server_list:
                    if server_info.get("server_name", "").lower() == cluster_id.lower():
                        found_servers.append(server_info)

            if found_servers:
                cluster = found_servers
            else:
                raise ValueError(
                    f"Кластер или сервер с ID/именем {cluster_id} не найден."
                )

        for server_info in cluster:
            panel_type = server_info.get("panel_type", "3x-ui").lower()
            server_name = server_info.get("server_name", "unknown")

            if panel_type == "remnawave":
                remna = RemnawaveAPI(server_info["api_url"])
                if not await remna.login(REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD):
                    logger.error(
                        f"[Remnawave] Не удалось войти на сервер {server_name}"
                    )
                    continue

                success = await remna.delete_user(client_id)
                if success:
                    logger.info(
                        f"[Remnawave] Клиент {client_id} успешно удалён с {server_name}"
                    )
                else:
                    logger.warning(
                        f"[Remnawave] Не удалось удалить клиента {client_id} с {server_name}"
                    )

            elif panel_type == "3x-ui":
                xui = await get_xui_instance(server_info["api_url"])
                inbound_id = server_info.get("inbound_id")

                if not inbound_id:
                    logger.warning(
                        f"[3x-ui] INBOUND_ID отсутствует на сервере {server_name}. Пропуск."
                    )
                    continue

                await delete_client(
                    xui,
                    inbound_id=int(inbound_id),
                    email=email,
                    client_id=client_id,
                )
                logger.info(
                    f"[3x-ui] Клиент {client_id} удалён с сервера {server_name}"
                )

            else:
                logger.warning(
                    f"[Unknown] Неизвестный тип панели '{panel_type}' для сервера {server_name}"
                )

    except Exception as e:
        logger.error(
            f"❌ Ошибка при удалении ключа {client_id} из кластера/сервера {cluster_id}: {e}"
        )
        raise


async def update_key_on_cluster(
    tg_id: int,
    client_id: str,
    email: str,
    expiry_time: int,
    cluster_id: str,
    session: AsyncSession,
):
    """
    Пересоздаёт ключ на всех серверах указанного кластера (или сервера, если передано имя).
    Работает с панелями 3x-ui и Remnawave. Возвращает кортеж: (новый client_id, remnawave ссылка или None).
    """
    try:
        servers = await get_servers(session)
        cluster = servers.get(cluster_id)

        if not cluster:
            found_servers = []
            for _key, server_list in servers.items():
                for server_info in server_list:
                    if server_info.get("server_name", "").lower() == cluster_id.lower():
                        found_servers.append(server_info)
            if found_servers:
                cluster = found_servers
            else:
                raise ValueError(
                    f"Кластер или сервер с ID/именем {cluster_id} не найден."
                )

        expire_iso = (
            datetime.utcfromtimestamp(expiry_time / 1000)
            .replace(tzinfo=timezone.utc)
            .isoformat()
        )

        remnawave_servers = [
            s for s in cluster if s.get("panel_type", "3x-ui").lower() == "remnawave"
        ]
        xui_servers = [
            s for s in cluster if s.get("panel_type", "3x-ui").lower() == "3x-ui"
        ]

        remnawave_client_id = None
        remnawave_key = None

        if remnawave_servers:
            inbound_ids = [
                s["inbound_id"] for s in remnawave_servers if s.get("inbound_id")
            ]
            remna = RemnawaveAPI(remnawave_servers[0]["api_url"])
            if await remna.login(REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD):
                await remna.delete_user(client_id)

                group_code = remnawave_servers[0].get("tariff_group")
                if not group_code:
                    raise ValueError("У Remnawave-сервера отсутствует tariff_group")

                result = await session.execute(
                    select(Tariff)
                    .where(Tariff.group_code == group_code, Tariff.is_active is True)
                    .order_by(Tariff.duration_days.desc())
                    .limit(1)
                )
                tariff = result.scalar_one_or_none()

                user_data = {
                    "username": email,
                    "trafficLimitStrategy": "NO_RESET",
                    "expireAt": expire_iso,
                    "telegramId": tg_id,
                    "activeUserInbounds": inbound_ids,
                }

                if tariff:
                    if tariff.traffic_limit is not None:
                        user_data["trafficLimitBytes"] = int(tariff.traffic_limit)
                    if tariff.device_limit is not None:
                        user_data["hwidDeviceLimit"] = int(tariff.device_limit)

                result = await remna.create_user(user_data)
                if result:
                    remnawave_client_id = result.get("uuid")
                    remnawave_key = result.get("subscriptionUrl")
                    logger.info(
                        f"[Update] Remnawave: клиент заново создан, новый UUID: {remnawave_client_id}"
                    )
                else:
                    logger.error("[Update] Ошибка создания Remnawave клиента")
            else:
                logger.error("[Update] Не удалось авторизоваться в Remnawave")

        if not remnawave_client_id:
            logger.warning(
                f"[Update] Remnawave client_id не получен. Используется исходный: {client_id}"
            )
            remnawave_client_id = client_id

        tasks = []
        for server_info in xui_servers:
            server_name = server_info.get("server_name", "unknown")
            inbound_id = server_info.get("inbound_id")

            if not inbound_id:
                logger.warning(
                    f"[Update] INBOUND_ID отсутствует для сервера {server_name}. Пропуск."
                )
                continue

            xui = await get_xui_instance(server_info["api_url"])

            sub_id = email
            unique_email = f"{email}_{server_name.lower()}" if SUPERNODE else email

            group_code = server_info.get("tariff_group")
            if not group_code:
                raise ValueError(f"У сервера {server_name} отсутствует tariff_group")

            result = await session.execute(
                select(Tariff)
                .where(Tariff.group_code == group_code, Tariff.is_active is True)
                .order_by(Tariff.duration_days.desc())
                .limit(1)
            )
            tariff = result.scalar_one_or_none()

            total_gb_bytes = (
                int(tariff.traffic_limit) if tariff and tariff.traffic_limit else 0
            )
            device_limit = (
                int(tariff.device_limit)
                if tariff and tariff.device_limit is not None
                else None
            )

            config = ClientConfig(
                client_id=remnawave_client_id,
                email=unique_email,
                tg_id=tg_id,
                limit_ip=device_limit,
                total_gb=total_gb_bytes,
                expiry_time=expiry_time,
                enable=True,
                flow="xtls-rprx-vision",
                inbound_id=int(inbound_id),
                sub_id=sub_id,
            )

            tasks.append(add_client(xui, config))

        if tasks:
            await asyncio.gather(*tasks, return_exceptions=True)

        logger.info(
            f"[Update] Ключ {remnawave_client_id} обновлён на всех серверах кластера {cluster_id}"
        )
        return remnawave_client_id, remnawave_key

    except Exception as e:
        logger.error(
            f"[Update Error] Ошибка при обновлении ключа {client_id} на {cluster_id}: {e}"
        )
        raise


async def update_subscription(
    tg_id: int,
    email: str,
    session: AsyncSession,
    cluster_override: str = None,
    country_override: str = None,
) -> None:
    result = await session.execute(
        select(Key).where(Key.tg_id == tg_id, Key.email == email)
    )
    record = result.scalar_one_or_none()

    if not record:
        raise ValueError(f"The key {email} does not exist in database")

    expiry_time = record.expiry_time
    client_id = record.client_id
    old_cluster_id = record.server_id
    public_link = f"{PUBLIC_LINK}{email}/{tg_id}"

    await delete_key_from_cluster(old_cluster_id, email, client_id, session=session)

    await session.execute(delete(Key).where(Key.tg_id == tg_id, Key.email == email))
    await session.commit()

    new_cluster_id = (
        country_override or cluster_override or await get_least_loaded_cluster(session)
    )

    new_client_id, remnawave_key = await update_key_on_cluster(
        tg_id=tg_id,
        client_id=client_id,
        email=email,
        expiry_time=expiry_time,
        cluster_id=new_cluster_id,
        session=session,
    )

    servers = await get_servers(session)
    cluster_servers = servers.get(new_cluster_id, [])
    has_xui = any(s.get("panel_type", "").lower() == "3x-ui" for s in cluster_servers)

    final_key_link = public_link if has_xui else None

    await store_key(
        session=session,
        tg_id=tg_id,
        client_id=new_client_id,
        email=email,
        expiry_time=expiry_time,
        key=final_key_link,
        remnawave_link=remnawave_key,
        server_id=new_cluster_id,
    )


async def get_user_traffic(
    session: AsyncSession, tg_id: int, email: str
) -> dict[str, Any]:
    """
    Получает трафик пользователя на всех серверах, где у него есть ключ (3x-ui и Remnawave).
    Для Remnawave трафик считается один раз и отображается как "Remnawave (общий):".
    """
    result = await session.execute(
        select(Key.client_id, Key.server_id).where(
            Key.tg_id == tg_id, Key.email == email
        )
    )
    rows = result.all()
    if not rows:
        return {"status": "error", "message": "У пользователя нет активных ключей."}

    server_ids = {row.server_id for row in rows}

    result = await session.execute(
        select(Server).where(
            Server.server_name.in_(server_ids) | Server.cluster_name.in_(server_ids)
        )
    )
    server_rows = result.scalars().all()
    if not server_rows:
        logger.error(f"Не найдено серверов для: {server_ids}")
        return {
            "status": "error",
            "message": f"Серверы не найдены: {', '.join(server_ids)}",
        }

    servers_map = {
        s.server_name: {
            "server_name": s.server_name,
            "cluster_name": s.cluster_name,
            "api_url": s.api_url,
            "panel_type": s.panel_type,
        }
        for s in server_rows
    }

    user_traffic_data = {}
    tasks = []

    remnawave_client_id = None
    remnawave_checked = False
    remnawave_api_url = None

    async def fetch_traffic(server_info: dict, client_id: str) -> tuple[str, Any]:
        server_name = server_info["server_name"]
        api_url = server_info["api_url"]
        panel_type = server_info.get("panel_type", "3x-ui").lower()

        try:
            if panel_type == "3x-ui":
                xui = await get_xui_instance(api_url)
                traffic_info = await get_client_traffic(xui, client_id)
                if traffic_info["status"] == "success" and traffic_info["traffic"]:
                    client_data = traffic_info["traffic"][0]
                    used_gb = (client_data.up + client_data.down) / 1073741824
                    return server_name, round(used_gb, 2)
                else:
                    return server_name, "Ошибка получения трафика"
            else:
                return server_name, f"Неизвестная панель: {panel_type}"
        except Exception as e:
            return server_name, f"Ошибка: {e}"

    for row in rows:
        client_id = row.client_id
        server_id = row.server_id

        matched_servers = [
            s
            for s in servers_map.values()
            if s["server_name"] == server_id or s["cluster_name"] == server_id
        ]
        for server_info in matched_servers:
            panel_type = server_info.get("panel_type", "3x-ui").lower()

            if panel_type == "remnawave" and not remnawave_checked:
                remnawave_client_id = client_id
                remnawave_api_url = server_info["api_url"]
                remnawave_checked = True
            elif panel_type == "3x-ui":
                tasks.append(fetch_traffic(server_info, client_id))

    results = await asyncio.gather(*tasks, return_exceptions=True)
    for server, result in results:
        user_traffic_data[server] = result

    if remnawave_client_id and remnawave_api_url:
        try:
            remna = RemnawaveAPI(remnawave_api_url)
            if not await remna.login(REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD):
                user_traffic_data["Remnawave (общий)"] = "Не удалось авторизоваться"
            else:
                user_data = await remna.get_user_by_uuid(remnawave_client_id)
                if not user_data:
                    user_traffic_data["Remnawave (общий)"] = "Клиент не найден"
                else:
                    used_bytes = user_data.get("usedTrafficBytes", 0)
                    used_gb = round(used_bytes / 1073741824, 2)
                    user_traffic_data["Remnawave (общий)"] = used_gb
        except Exception as e:
            user_traffic_data["Remnawave (общий)"] = f"Ошибка: {e}"

    return {"status": "success", "traffic": user_traffic_data}


async def toggle_client_on_cluster(
    cluster_id: str,
    email: str,
    client_id: str,
    enable: bool = True,
    session: AsyncSession = None,
) -> dict[str, Any]:
    try:
        if session is None:
            raise ValueError(
                "[Cluster Toggle] Не передан объект сессии для toggle_client_on_cluster"
            )
        servers = await get_servers(session)
        cluster = servers.get(cluster_id)

        if not cluster:
            found_servers = []
            for _, server_list in servers.items():
                for server_info in server_list:
                    if server_info.get("server_name", "").lower() == cluster_id.lower():
                        found_servers.append(server_info)
            if found_servers:
                cluster = found_servers
            else:
                raise ValueError(
                    f"Кластер или сервер с ID/именем '{cluster_id}' не найден."
                )

        results = {}
        tasks = []

        for server_info in cluster:
            panel_type = server_info.get("panel_type", "3x-ui").lower()
            server_name = server_info.get("server_name", "unknown")

            if panel_type == "3x-ui":
                inbound_id = server_info.get("inbound_id")
                if not inbound_id:
                    logger.warning(
                        f"[3x-ui] INBOUND_ID отсутствует для сервера {server_name}. Пропуск."
                    )
                    results[server_name] = False
                    continue

                xui = await get_xui_instance(server_info["api_url"])
                unique_email = f"{email}_{server_name.lower()}" if SUPERNODE else email

                tasks.append(
                    toggle_client(xui, int(inbound_id), unique_email, client_id, enable)
                )

            elif panel_type == "remnawave":
                remna = RemnawaveAPI(server_info["api_url"])
                if not await remna.login(REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD):
                    logger.error(
                        f"[Remnawave] Авторизация не удалась на сервере {server_name}"
                    )
                    results[server_name] = False
                    continue

                func = remna.enable_user if enable else remna.disable_user
                tasks.append(func(client_id))

            else:
                logger.warning(
                    f"[Cluster Toggle] Неизвестный тип панели '{panel_type}' на сервере {server_name}. Пропуск."
                )
                results[server_name] = False

        task_results = await asyncio.gather(*tasks, return_exceptions=True)

        for server_info, result in zip(cluster, task_results, strict=False):
            server_name = server_info.get("server_name", "unknown")
            if isinstance(result, Exception):
                logger.error(
                    f"[Cluster Toggle] Ошибка на сервере {server_name}: {result}"
                )
                results[server_name] = False
            else:
                results[server_name] = result

        status = "включен" if enable else "отключен"
        logger.info(
            f"[Cluster Toggle] Клиент {email} {status} на серверах кластера {cluster_id}"
        )
        logger.info(f"[Cluster Toggle DEBUG] Результаты: {results}")

        return {
            "status": "success" if any(results.values()) else "error",
            "results": results,
        }

    except Exception as e:
        logger.error(
            f"[Cluster Toggle] Ошибка при изменении состояния клиента {email} в кластере {cluster_id}: {e}"
        )
        return {"status": "error", "error": str(e)}


async def reset_traffic_in_cluster(
    cluster_id: str, email: str, session: AsyncSession
) -> None:
    try:
        servers = await get_servers(session)
        cluster = servers.get(cluster_id)

        if not cluster:
            found_servers = []
            for _, server_list in servers.items():
                for server_info in server_list:
                    if server_info.get("server_name", "").lower() == cluster_id.lower():
                        found_servers.append(server_info)
            if found_servers:
                cluster = found_servers
            else:
                raise ValueError(
                    f"Кластер или сервер с ID/именем {cluster_id} не найден."
                )

        tasks = []
        remnawave_done = False

        for server_info in cluster:
            panel_type = server_info.get("panel_type", "3x-ui").lower()
            server_name = server_info.get("server_name", "unknown")
            api_url = server_info.get("api_url")
            inbound_id = server_info.get("inbound_id")

            if panel_type == "remnawave" and not remnawave_done:
                result = await session.execute(
                    select(Key.client_id)
                    .where(Key.email == email, Key.server_id == cluster_id)
                    .limit(1)
                )
                row = result.first()

                if not row:
                    logger.warning(
                        f"[Remnawave Reset] client_id не найден для {email} на {server_name}"
                    )
                    continue

                client_id = row[0]

                remna = RemnawaveAPI(api_url)
                if not await remna.login(REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD):
                    logger.warning(
                        f"[Reset Traffic] Не удалось авторизоваться в Remnawave ({server_name})"
                    )
                    continue

                tasks.append(remna.reset_user_traffic(client_id))
                remnawave_done = True
                continue

            if panel_type == "3x-ui":
                if not inbound_id:
                    logger.warning(
                        f"INBOUND_ID отсутствует для сервера {server_name}. Пропуск."
                    )
                    continue

                xui = await get_xui_instance(api_url)
                unique_email = f"{email}_{server_name.lower()}" if SUPERNODE else email
                tasks.append(xui.client.reset_stats(int(inbound_id), unique_email))
            else:
                logger.warning(
                    f"[Reset Traffic] Неизвестный тип панели '{panel_type}' на {server_name}"
                )

        await asyncio.gather(*tasks, return_exceptions=True)
        logger.info(
            f"[Reset Traffic] Трафик клиента {email} успешно сброшен в кластере {cluster_id}"
        )

    except Exception as e:
        logger.error(
            f"[Reset Traffic] Ошибка при сбросе трафика клиента {email} в кластере {cluster_id}: {e}"
        )
        raise
