vpn/app/services/xui_panels.py

"""CRUD + connection tests for linked 3x-ui panels."""

from __future__ import annotations

import json
from datetime import datetime, timezone
from typing import Any

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

from app.models import XuiPanel, XuiPanelInbound, XuiSiteClient
from app.services.xui import (
    SUPPORTED_CLIENT_PROTOCOLS,
    XuiApiError,
    XuiClient,
    normalize_base_url,
)


def _client_for(panel: XuiPanel) -> XuiClient:
    return XuiClient(
        panel.base_url,
        username=panel.username,
        password=panel.password,
        api_token=panel.api_token,
        verify_ssl=panel.verify_ssl,
    )


async def list_panels(session: AsyncSession) -> list[XuiPanel]:
    result = await session.execute(select(XuiPanel).order_by(XuiPanel.id.desc()))
    return list(result.scalars().all())


async def create_panel(
    session: AsyncSession,
    *,
    name: str,
    base_url: str,
    username: str = "",
    password: str = "",
    api_token: str = "",
    verify_ssl: bool = True,
) -> XuiPanel:
    token = api_token.strip() or None
    user = username.strip() or None
    pwd = password.strip() or None
    if not token and not (user and pwd):
        raise ValueError("Укажите API Token или логин и пароль")

    url = normalize_base_url(base_url)
    panel = XuiPanel(
        name=(name.strip() or url),
        base_url=url,
        username=user,
        password=pwd,
        api_token=token,
        verify_ssl=verify_ssl,
        is_enabled=True,
    )
    session.add(panel)
    await session.commit()
    await session.refresh(panel)

    await check_panel(session, panel.id)
    await session.refresh(panel)
    if not panel.is_reachable:
        raise ValueError(panel.last_error or "Не удалось подключиться к 3x-ui")
    return panel


async def check_panel(session: AsyncSession, panel_id: int) -> XuiPanel:
    panel = await session.get(XuiPanel, panel_id)
    if not panel:
        raise ValueError("Панель не найдена")

    now = datetime.now(timezone.utc)
    try:
        async with _client_for(panel) as client:
            info = await client.test_connection()
        panel.is_reachable = True
        panel.last_error = None
        panel.panel_info = json.dumps(info, ensure_ascii=False, default=str)[:8000]
        panel.last_checked = now
    except (XuiApiError, ValueError, OSError) as exc:
        panel.is_reachable = False
        panel.last_error = str(exc)[:2000]
        panel.last_checked = now
        await session.commit()
        await session.refresh(panel)
        raise ValueError(str(exc)) from exc

    await session.commit()
    await session.refresh(panel)
    try:
        await sync_panel_inbounds(session, panel_id)
    except Exception:  # noqa: BLE001
        pass
    await session.refresh(panel)
    return panel


async def delete_panel(session: AsyncSession, panel_id: int) -> None:
    panel = await session.get(XuiPanel, panel_id)
    if panel:
        await session.delete(panel)
        await session.commit()


async def fetch_panel_data(panel: XuiPanel) -> dict[str, Any]:
    async with _client_for(panel) as client:
        status = await client.get_status()
        all_inbounds = await client.list_inbounds()
        available = await client.list_available_inbounds(
            protocols=SUPPORTED_CLIENT_PROTOCOLS, enabled_only=False
        )
        vless_inbounds = [i for i in available if str(i.get("protocol")).lower() == "vless"]
        wg_inbounds = [i for i in available if str(i.get("protocol")).lower() == "wireguard"]
        try:
            clients = await client.list_clients()
        except XuiApiError:
            clients = []
        return {
            "status": status,
            "inbounds": all_inbounds,
            "available_inbounds": available,
            "vless_inbounds": vless_inbounds,
            "wireguard_inbounds": wg_inbounds,
            "clients": clients,
        }


async def load_available_inbounds(
    panel: XuiPanel,
    *,
    protocol: str | None = None,
) -> list[dict]:
    protocols: tuple[str, ...]
    if protocol:
        protocols = (protocol.lower(),)
    else:
        protocols = SUPPORTED_CLIENT_PROTOCOLS
    async with _client_for(panel) as client:
        return await client.list_available_inbounds(protocols=protocols, enabled_only=True)


async def sync_panel_inbounds(session: AsyncSession, panel_id: int) -> list[XuiPanelInbound]:
    """Fetch VLESS/WG inbounds from 3x-ui and upsert into DB (preserve user flags)."""
    panel = await session.get(XuiPanel, panel_id)
    if not panel:
        raise ValueError("Панель не найдена")

    remote = await load_available_inbounds(panel)
    now = datetime.now(timezone.utc)
    seen_ids: set[int] = set()

    existing_rows = (
        await session.execute(select(XuiPanelInbound).where(XuiPanelInbound.panel_id == panel_id))
    ).scalars().all()
    by_inbound = {r.inbound_id: r for r in existing_rows}

    for item in remote:
        inbound_id = int(item.get("id"))
        seen_ids.add(inbound_id)
        row = by_inbound.get(inbound_id)
        if row is None:
            row = XuiPanelInbound(
                panel_id=panel_id,
                inbound_id=inbound_id,
                is_available_for_users=False,
            )
            session.add(row)
        row.protocol = str(item.get("protocol") or "").lower()
        row.remark = item.get("remark")
        port = item.get("port")
        row.port = int(port) if port is not None else None
        row.enable = bool(item.get("enable", True))
        row.last_synced_at = now

    for row in existing_rows:
        if row.inbound_id not in seen_ids:
            await session.delete(row)

    await session.commit()
    return await list_panel_inbounds(session, panel_id)


async def list_panel_inbounds(session: AsyncSession, panel_id: int) -> list[XuiPanelInbound]:
    result = await session.execute(
        select(XuiPanelInbound)
        .where(XuiPanelInbound.panel_id == panel_id)
        .order_by(XuiPanelInbound.protocol, XuiPanelInbound.inbound_id)
    )
    return list(result.scalars().all())


async def list_user_inbounds(
    session: AsyncSession,
    panel_id: int,
    *,
    protocol: str | None = None,
) -> list[XuiPanelInbound]:
    query = select(XuiPanelInbound).where(
        XuiPanelInbound.panel_id == panel_id,
        XuiPanelInbound.is_available_for_users.is_(True),
        XuiPanelInbound.enable.is_(True),
    )
    if protocol:
        query = query.where(XuiPanelInbound.protocol == protocol.lower())
    query = query.order_by(XuiPanelInbound.protocol, XuiPanelInbound.inbound_id)
    result = await session.execute(query)
    return list(result.scalars().all())


async def set_inbound_for_users(
    session: AsyncSession,
    panel_id: int,
    inbound_id: int,
    *,
    available: bool,
) -> XuiPanelInbound:
    result = await session.execute(
        select(XuiPanelInbound).where(
            XuiPanelInbound.panel_id == panel_id,
            XuiPanelInbound.inbound_id == inbound_id,
        )
    )
    row = result.scalar_one_or_none()
    if not row:
        raise ValueError("Inbound не найден — сначала синхронизируйте список")
    row.is_available_for_users = available
    await session.commit()
    await session.refresh(row)
    return row


async def save_user_inbounds(
    session: AsyncSession,
    panel_id: int,
    inbound_ids: list[int],
) -> list[XuiPanelInbound]:
    """Set which inbound IDs are available for users (others off)."""
    rows = await list_panel_inbounds(session, panel_id)
    selected = set(inbound_ids)
    for row in rows:
        row.is_available_for_users = row.inbound_id in selected
    await session.commit()
    return await list_panel_inbounds(session, panel_id)


def compute_expiry_time(*, days: int, start_after_first_use: bool) -> int:
    """
    3x-ui expiryTime encoding:
      0  — unlimited
      >0 — absolute deadline in unix ms
      <0 — duration ms; countdown starts after first use
    """
    days = int(days or 0)
    if days <= 0:
        return 0
    duration_ms = days * 24 * 60 * 60 * 1000
    if start_after_first_use:
        return -duration_ms
    import time

    return int(time.time() * 1000) + duration_ms


async def list_site_clients(session: AsyncSession, panel_id: int) -> list[XuiSiteClient]:
    result = await session.execute(
        select(XuiSiteClient)
        .where(XuiSiteClient.panel_id == panel_id)
        .order_by(XuiSiteClient.id.desc())
    )
    return list(result.scalars().all())


async def list_all_site_clients(session: AsyncSession) -> list[XuiSiteClient]:
    from sqlalchemy.orm import selectinload

    result = await session.execute(
        select(XuiSiteClient)
        .options(selectinload(XuiSiteClient.panel))
        .order_by(XuiSiteClient.id.desc())
    )
    return list(result.scalars().all())


async def add_xui_client(
    session: AsyncSession,
    panel: XuiPanel,
    *,
    protocol: str,
    email: str,
    inbound_id: int,
    total_gb: int = 0,
    limit_ip: int = 0,
    expiry_days: int = 0,
    start_after_first_use: bool = False,
    flow: str = "",
    comment: str = "",
    require_user_available: bool = False,
) -> dict:
    proto = protocol.strip().lower()
    if require_user_available:
        allowed = await list_user_inbounds(session, panel.id, protocol=proto)
        if inbound_id not in {r.inbound_id for r in allowed}:
            raise ValueError("Этот inbound недоступен для пользователей — включите его на сервере 3x-ui")

    bytes_limit = total_gb * 1024 * 1024 * 1024 if total_gb > 0 else 0
    expiry_time = compute_expiry_time(days=expiry_days, start_after_first_use=start_after_first_use)
    async with _client_for(panel) as client:
        if proto == "vless":
            result = await client.add_vless_client(
                email=email,
                inbound_id=inbound_id,
                total_gb=bytes_limit,
                expiry_time=expiry_time,
                limit_ip=limit_ip,
                flow=flow,
                comment=comment,
            )
        elif proto == "wireguard":
            result = await client.add_wireguard_client(
                email=email,
                inbound_id=inbound_id,
                total_gb=bytes_limit,
                expiry_time=expiry_time,
                limit_ip=limit_ip,
                comment=comment,
            )
        else:
            raise ValueError("Поддерживаются только vless и wireguard")

    links = result.get("links") or []
    inbound_meta = result.get("inbound") or {}
    # Prefer remark from loaded inbound options if missing
    remark = inbound_meta.get("remark")
    if not remark and proto == "vless":
        remark = f"vless#{inbound_id}"

    existing = await session.execute(
        select(XuiSiteClient).where(
            XuiSiteClient.panel_id == panel.id,
            XuiSiteClient.email == email,
        )
    )
    row = existing.scalar_one_or_none()
    if row is None:
        row = XuiSiteClient(panel_id=panel.id, email=email)
        session.add(row)

    row.protocol = proto
    row.inbound_id = inbound_id
    row.inbound_remark = remark
    row.uuid = result.get("uuid")
    row.flow = result.get("flow")
    row.private_key = result.get("privateKey")
    row.public_key = result.get("publicKey")
    row.link = links[0] if links else None
    row.expiry_days = expiry_days
    row.start_after_first_use = start_after_first_use
    row.expiry_time = int(result.get("expiryTime") or expiry_time or 0)
    row.total_gb = total_gb
    row.limit_ip = limit_ip
    row.comment = comment or None
    row.created_via = "website"
    await session.commit()
    await session.refresh(row)

    result["expiry_days"] = expiry_days
    result["start_after_first_use"] = start_after_first_use
    result["site_client_id"] = row.id
    result["created_via"] = "website"
    return result