"""Application-level shared aiohttp.ClientSession.

All outgoing HTTP requests in the server package should use the shared
session returned by ``get_http_session()`` instead of creating
per-request ``aiohttp.ClientSession`` instances.  This keeps a single
TCP connection pool alive for the lifetime of the process, avoiding
the overhead of pool creation/teardown on every request.

DNS-rebinding mitigation
~~~~~~~~~~~~~~~~~~~~~~~~
The session is wired with a :class:`PinnedResolver` from
``notify_bridge_core.notifications.ssrf`` so the IP that passed the
SSRF block-range check during URL validation is the one aiohttp
actually connects to. Without this pinning a malicious DNS server
could swap a public IP for ``127.0.0.1`` between validation and
connect, defeating the SSRF guard.

Callers that do their own ``avalidate_outbound_url_full`` should also
call :func:`pin_validated` to register the resolved host->IP mapping
on the shared resolver before issuing the request. Callers that just
use the session opportunistically still benefit from scheme + range
checks at validation sites, plus the fallback resolver here.

Call ``close_http_session()`` once during application shutdown.
"""

from __future__ import annotations

import asyncio

import aiohttp

from notify_bridge_core.notifications.ssrf import PinnedResolver, ValidatedURL

_DEFAULT_TIMEOUT = aiohttp.ClientTimeout(total=30, connect=10)

_session: aiohttp.ClientSession | None = None
_resolver: PinnedResolver | None = None
# Lazy init: ``asyncio.Lock()`` at module import time binds to whichever
# event loop happens to be running (or none). Tests that spin up multiple
# loops (or subprocesses with their own loop) would otherwise hit
# "RuntimeError: ... attached to a different loop". Defer creation to
# first use so the lock binds to the loop that actually calls us.
_lock: asyncio.Lock | None = None


def _get_lock() -> asyncio.Lock:
    """Return the module lock, creating it on first call from this loop."""
    global _lock
    if _lock is None:
        _lock = asyncio.Lock()
    return _lock


async def get_http_session() -> aiohttp.ClientSession:
    """Get or create the shared HTTP session.

    Concurrent first-callers are serialized through the lazy lock so we
    never leak a second ClientSession / connector pair. Once established,
    hot callers skip the lock via the fast-path check.

    The session uses a :class:`PinnedResolver` connector so callers that
    register validated host->IP mappings via :func:`pin_validated` defeat
    DNS rebinding between validation and connect.
    """
    global _session, _resolver
    if _session is not None and not _session.closed:
        return _session
    async with _get_lock():
        if _session is None or _session.closed:
            _resolver = PinnedResolver()
            connector = aiohttp.TCPConnector(resolver=_resolver)
            _session = aiohttp.ClientSession(
                timeout=_DEFAULT_TIMEOUT, connector=connector,
            )
    return _session


def pin_validated(validated: ValidatedURL) -> None:
    """Register a validated (host, ip) mapping on the shared resolver.

    Best-effort: if the resolver has not been created yet (no session
    initialised), the call is a no-op. Once the session exists, every
    aiohttp connect for ``validated.host`` will use ``validated.ip``.
    """
    if _resolver is None:
        return
    _resolver.pin(validated.host, validated.ip)


async def close_http_session() -> None:
    """Close the shared HTTP session (call on app shutdown)."""
    global _session, _resolver
    async with _get_lock():
        if _session is not None and not _session.closed:
            await _session.close()
        if _resolver is not None:
            await _resolver.close()
        _session = None
        _resolver = None
