"""Per-invocation async session factory for Celery task modules.

Async engine connections are bound to the event loop. Each Celery task
runs its async body under a fresh ``asyncio.run()`` loop, so it needs its
own engine created (and disposed) within that loop — a process-wide async
engine would reuse loop-bound connections across tasks and raise "attached
to a different loop". So unlike the process-wide sync engine in
``_sync_engine.py``, this returns a fresh engine per call; the caller
disposes it (``await engine.dispose()``) when its loop ends.
"""

from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.pool import NullPool

from ..config import get_config


def async_session_factory(*, server_settings: dict | None = None):
    """Return ``(sessionmaker, engine)`` bound to a fresh async engine.

    ``server_settings`` (optional) are applied by asyncpg as per-connection GUCs
    on connect. Because NullPool opens a fresh real connection per checkout, every
    transaction in the task inherits them — used to set ``lock_timeout`` so a
    statement blocked on a lock fails fast instead of hanging to the Celery hard
    limit (operator-flagged normalize_tags timeout 2026-06-07).
    """
    cfg = get_config()
    # NullPool: this engine lives for ONE task (created + disposed per
    # asyncio.run loop), so intra-task connection pooling buys nothing and
    # actively bit us — download_source releases its phase-1 connection
    # before a multi-minute gallery-dl subprocess, and a *pooled* idle
    # connection would be reaped by the server and handed back dead to
    # phase 3 (asyncpg ConnectionDoesNotExistError, Anduo #40014). NullPool
    # opens a fresh real connection on each checkout, so phase 3 always
    # reconnects clean; pre_ping is then redundant.
    connect_args = {}
    if server_settings:
        connect_args["server_settings"] = {
            k: str(v) for k, v in server_settings.items()
        }
    engine = create_async_engine(
        cfg.database_url, future=True, poolclass=NullPool,
        connect_args=connect_args,
    )
    return async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False), engine