BanGUI/backend/app/services/geo_service.py

"""Geo service — backward compatibility wrappers.

DEPRECATED: This module is kept for backward compatibility only. New code should
use :class:`GeoCache` directly from ``app.services.geo_cache``. The underlying
implementation has been refactored to eliminate module-level mutable state.

The :class:`GeoCache` instance should be injected into services and routers
via dependency injection, and stored on ``app.state.geo_cache``.

See :class:`app.services.geo_cache.GeoCache` for the implementation.
"""

from __future__ import annotations

from typing import TYPE_CHECKING

from app.services.geo_cache import GeoCache
from app.models.geo import GeoInfo

if TYPE_CHECKING:
    import aiohttp
    import aiosqlite

__all__ = [
    "GeoCache",
    "clear_cache",
    "clear_neg_cache",
    "is_cached",
    "lookup",
    "lookup_batch",
    "lookup_cached_only",
    "cache_stats",
    "count_unresolved",
    "get_unresolved_ips",
    "load_cache_from_db",
    "flush_dirty",
    "re_resolve_all",
    "init_geoip",
]

#: Deprecated: Module-level cache instance for backward compatibility.
#: This exists only to provide backward-compatible module-level functions.
#: New code should inject GeoCache instances directly.
_default_geo_cache: GeoCache = GeoCache()


async def clear_cache() -> None:
    """(DEPRECATED) Flush both the positive and negative lookup caches.

    Use :meth:`GeoCache.clear` instead. This function delegates to the
    default module-level instance for backward compatibility only.
    """
    await _default_geo_cache.clear()


async def clear_neg_cache() -> None:
    """(DEPRECATED) Flush only the negative (failed-lookups) cache.

    Use :meth:`GeoCache.clear_neg_cache` instead. This function delegates to
    the default module-level instance for backward compatibility only.
    """
    await _default_geo_cache.clear_neg_cache()


def is_cached(ip: str) -> bool:
    """(DEPRECATED) Return ``True`` if *ip* has a positive cache entry.

    Use :meth:`GeoCache.is_cached` instead. This function delegates to the
    default module-level instance for backward compatibility only.

    Args:
        ip: IPv4 or IPv6 address string.

    Returns:
        ``True`` when *ip* is in the cache with a known country code.
    """
    return _default_geo_cache.is_cached(ip)


async def cache_stats(db: aiosqlite.Connection) -> dict[str, int]:
    """(DEPRECATED) Return diagnostic counters for the geo cache subsystem.

    Use :meth:`GeoCache.cache_stats` instead. This function delegates to the
    default module-level instance for backward compatibility only.

    Args:
        db: Open BanGUI application database connection.

    Returns:
        Dict with keys ``cache_size``, ``unresolved``, ``neg_cache_size``,
        and ``dirty_size``.
    """
    return await _default_geo_cache.cache_stats(db)


async def count_unresolved(db: aiosqlite.Connection) -> int:
    """(DEPRECATED) Return the number of unresolved entries in the geo cache.

    Use :meth:`GeoCache.count_unresolved` instead.
    """
    return await _default_geo_cache.count_unresolved(db)


async def get_unresolved_ips(db: aiosqlite.Connection) -> list[str]:
    """(DEPRECATED) Return IPs with NULL country_code in the persistent cache.

    Use :meth:`GeoCache.get_unresolved_ips` instead.

    Args:
        db: Open BanGUI application database connection.

    Returns:
        List of IP addresses that are candidates for re-resolution.
    """
    return await _default_geo_cache.get_unresolved_ips(db)


def init_geoip(mmdb_path: str | None) -> None:
    """(DEPRECATED) Initialise the MaxMind GeoLite2-Country database reader.

    Use :meth:`GeoCache.init_geoip` instead. This function delegates to the
    default module-level instance for backward compatibility only.

    Args:
        mmdb_path: Absolute path to a ``GeoLite2-Country.mmdb`` file.
    """
    _default_geo_cache.init_geoip(mmdb_path)


async def load_cache_from_db(db: aiosqlite.Connection) -> None:
    """(DEPRECATED) Pre-populate the in-memory cache from ``geo_cache`` table.

    Use :meth:`GeoCache.load_cache_from_db` instead. This function delegates
    to the default module-level instance for backward compatibility only.

    Args:
        db: Open :class:`aiosqlite.Connection` to the BanGUI application
            database (not the fail2ban database).
    """
    await _default_geo_cache.load_cache_from_db(db)


async def lookup(
    ip: str,
    http_session: aiohttp.ClientSession,
    db: aiosqlite.Connection | None = None,
) -> GeoInfo | None:
    """(DEPRECATED) Resolve an IP address to country, ASN, organisation metadata.

    Use :meth:`GeoCache.lookup` instead. This function delegates to the
    default module-level instance for backward compatibility only.

    Args:
        ip: IPv4 or IPv6 address string.
        http_session: Shared :class:`aiohttp.ClientSession`.
        db: Optional BanGUI application database for persistence.

    Returns:
        A :class:`GeoInfo` instance, or ``None`` on lookup failure.
    """
    return await _default_geo_cache.lookup(ip, http_session, db=db)


def lookup_cached_only(
    ips: list[str],
) -> tuple[dict[str, GeoInfo], list[str]]:
    """(DEPRECATED) Return cached geo data without making external API calls.

    Use :meth:`GeoCache.lookup_cached_only` instead. This function delegates
    to the default module-level instance for backward compatibility only.

    Args:
        ips: IP address strings to look up.

    Returns:
        A ``(geo_map, uncached)`` tuple.
    """
    return _default_geo_cache.lookup_cached_only(ips)


async def lookup_batch(
    ips: list[str],
    http_session: aiohttp.ClientSession,
    db: aiosqlite.Connection | None = None,
) -> dict[str, GeoInfo]:
    """(DEPRECATED) Resolve multiple IPs in bulk using ip-api.com batch endpoint.

    Use :meth:`GeoCache.lookup_batch` instead. This function delegates to the
    default module-level instance for backward compatibility only.

    Args:
        ips: List of IP address strings to resolve.
        http_session: Shared :class:`aiohttp.ClientSession`.
        db: Optional BanGUI application database for persistent cache writes.

    Returns:
        Dict mapping ``ip → GeoInfo`` for every input IP.
    """
    return await _default_geo_cache.lookup_batch(ips, http_session, db=db)


async def flush_dirty(db: aiosqlite.Connection) -> int:
    """(DEPRECATED) Persist all new in-memory geo entries to the database.

    Use :meth:`GeoCache.flush_dirty` instead. This function delegates to the
    default module-level instance for backward compatibility only.

    Args:
        db: Open :class:`aiosqlite.Connection` to the BanGUI application
            database.

    Returns:
        The number of rows successfully upserted.
    """
    return await _default_geo_cache.flush_dirty(db)


async def re_resolve_all(
    db: aiosqlite.Connection,
    http_session: aiohttp.ClientSession,
) -> dict[str, int]:
    """(DEPRECATED) Retry geo resolution for all unresolved cache entries.

    Use :meth:`GeoCache.re_resolve_all` instead. This function delegates to
    the default module-level instance for backward compatibility only.

    Args:
        db: BanGUI application database connection.
        http_session: Shared aiohttp client session.

    Returns:
        A dict with ``resolved`` and ``total`` counts.
    """
    return await _default_geo_cache.re_resolve_all(db, http_session)