bike-app/apps/api/velodrome/db.py

"""Two database engines, on purpose.

`app` engine: connects as a role with RLS enforced (NOBYPASSRLS). Every request handler that has
already established who the caller is uses this, wrapped in `scoped_session()` below, which sets
`app.user_id` for the transaction so RLS policies can key off it.

`auth` engine: connects as a role with BYPASSRLS. Used ONLY by velodrome.auth.service, and only
for the narrow set of queries that must run *before* identity is known — looking a user up by
email at login, a session up by its hashed token, an invite up by its hashed code — plus the
inserts that create those rows in the first place. Nothing outside auth/service.py should import
this engine; if you find yourself reaching for it elsewhere, the query almost certainly belongs
in a repository method on the scoped session instead (see CLAUDE.md invariant #4).
"""

from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from uuid import UUID

from sqlalchemy import text
from sqlalchemy.ext.asyncio import (
    AsyncEngine,
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)
from sqlalchemy.pool import NullPool

from velodrome.config import get_settings


def _make_engine(url: str) -> AsyncEngine:
    settings = get_settings()
    # NullPool in dev/test keeps behaviour predictable across the pytest-asyncio event loop;
    # production tuning (pool_size etc.) is a deploy-time concern, not a Phase-0 one.
    return create_async_engine(
        url,
        poolclass=NullPool if settings.environment == "test" else None,
        echo=False,
    )


_app_engine: AsyncEngine | None = None
_auth_engine: AsyncEngine | None = None


def app_engine() -> AsyncEngine:
    global _app_engine
    if _app_engine is None:
        _app_engine = _make_engine(str(get_settings().database_url_app))
    return _app_engine


def auth_engine() -> AsyncEngine:
    global _auth_engine
    if _auth_engine is None:
        _auth_engine = _make_engine(str(get_settings().database_url_auth))
    return _auth_engine


_app_sessionmaker: async_sessionmaker[AsyncSession] | None = None
_auth_sessionmaker: async_sessionmaker[AsyncSession] | None = None


def _app_sessions() -> async_sessionmaker[AsyncSession]:
    global _app_sessionmaker
    if _app_sessionmaker is None:
        _app_sessionmaker = async_sessionmaker(app_engine(), expire_on_commit=False)
    return _app_sessionmaker


def _auth_sessions() -> async_sessionmaker[AsyncSession]:
    global _auth_sessionmaker
    if _auth_sessionmaker is None:
        _auth_sessionmaker = async_sessionmaker(auth_engine(), expire_on_commit=False)
    return _auth_sessionmaker


@asynccontextmanager
async def auth_session() -> AsyncIterator[AsyncSession]:
    """A BYPASSRLS session. See module docstring — auth/service.py only."""
    async with _auth_sessions()() as session:
        yield session


@asynccontextmanager
async def scoped_session(user_id: UUID) -> AsyncIterator[AsyncSession]:
    """An RLS-scoped session for a known, authenticated user.

    `SET LOCAL` binds to the current transaction, not the connection, so this is safe under
    connection pooling — it can never leak `app.user_id` from one request into a pooled
    connection reused by a different request.
    """
    async with _app_sessions()() as session:
        async with session.begin():
            await session.execute(
                # bound parameter, not string interpolation — user_id is a UUID we generated
                # or validated ourselves, but there is no reason to ever risk it.
                text("SELECT set_config('app.user_id', :uid, true)"),
                {"uid": str(user_id)},
            )
            yield session


@asynccontextmanager
async def unscoped_session() -> AsyncIterator[AsyncSession]:
    """An `app`-role session with no `app.user_id` set.

    RLS policies default-deny when `current_setting('app.user_id', true)` is NULL, so this sees
    zero rows of any user-owned table — useful for health checks and anything that only touches
    non-RLS tables. Prefer `scoped_session` whenever a user is known.
    """
    async with _app_sessions()() as session:
        yield session