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

"""Single-engine SQLite access.

Phase 0 originally ran two Postgres roles with row-level security as a database-enforced
isolation layer (`velodrome_app`/`velodrome_auth` — see docs/DECISIONS.md D4). D15 moved the
database to SQLite, which has no roles, no session variables, and no policy engine — there is no
database-enforced layer left. Isolation between users now rests entirely on the query builder
below. CLAUDE.md's invariant #4 treats this file as load-bearing, not a convenience wrapper: a
new user-owned table without a passing isolation test (see tests/test_auth.py's pattern) is not
done.

Two access patterns:

- `scoped_session(user_id)` — for every query against a user-owned table once identity is known.
  Yields a `Scope`, whose `select()` is the ONLY way to build a query through it — every query it
  builds is pre-filtered to that user_id, on any model that declares a `user_id` column. There is
  no method on `Scope` that returns an unfiltered query. This is what makes "forgot the WHERE
  clause" structurally harder than remembering to write one by hand.
- `auth_session()` — for velodrome.auth.service ONLY: the handful of pre-identity lookups (login
  by email, a session by its token hash, an invite by its code hash) that by definition can't be
  scoped to a user_id nobody has established yet. Every query on this session must be an exact
  match on a unique key, never an unfiltered scan — that discipline is what made bypassing RLS
  safe before, and it's what keeps this safe now that RLS is gone. Nothing outside auth/service.py
  should import this.
"""

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

from sqlalchemy import Select, event, select
from sqlalchemy.ext.asyncio import (
    AsyncEngine,
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.pool import NullPool

from velodrome.config import get_settings


def _configure_sqlite_for_concurrent_writers(eng: AsyncEngine) -> None:
    """Two SQLite defaults that silently do the wrong thing if left alone — confirmed empirically
    against a real aiosqlite connection, not assumed from docs:

    1. Foreign key enforcement, `ON DELETE CASCADE` included, is OFF by default per connection.
       Without `PRAGMA foreign_keys=ON`, deleting a user leaves its sessions/api_tokens rows
       behind instead of cascading — verified this happens silently, no error either way.
    2. pysqlite/aiosqlite default to a DEFERRED transaction, which only takes a write lock on the
       first actual write statement — leaving a real check-then-act race window. Concretely: two
       concurrent redemptions of the same invite code could both read `used_count < max_uses` as
       true before either commits, double-spending a single-use invite. Disable the driver's own
       implicit BEGIN handling and issue `BEGIN IMMEDIATE` ourselves instead, which takes the
       write lock at transaction start and correctly serializes writers — SQLAlchemy's own
       documented recipe for this, not a workaround improvised here.
    """

    @event.listens_for(eng.sync_engine, "connect")
    def _on_connect(dbapi_connection: Any, connection_record: Any) -> None:
        cursor = dbapi_connection.cursor()
        cursor.execute("PRAGMA foreign_keys=ON")
        cursor.close()
        dbapi_connection.isolation_level = None

    @event.listens_for(eng.sync_engine, "begin")
    def _begin_immediate(conn: Any) -> None:
        conn.exec_driver_sql("BEGIN IMMEDIATE")


_engine: AsyncEngine | None = None
_sessionmaker: async_sessionmaker[AsyncSession] | None = None


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


def _sessions() -> async_sessionmaker[AsyncSession]:
    global _sessionmaker
    if _sessionmaker is None:
        _sessionmaker = async_sessionmaker(_engine_instance(), expire_on_commit=False)
    return _sessionmaker


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


@asynccontextmanager
async def unscoped_session() -> AsyncIterator[AsyncSession]:
    """A plain session with no user scoping applied at all — for health checks and anything that
    never touches a user-owned table. Prefer `scoped_session` whenever a user is known; reaching
    for this instead of that for a query against a user-owned table is exactly the mistake
    invariant #4 exists to catch in review.
    """
    async with _sessions()() as session:
        yield session


class Scope:
    """A user-scoped query builder. `select()` is the only way to build a query through this
    object, and it is always pre-filtered to `user_id` — there is no method here that returns an
    unfiltered query against a user-owned table.
    """

    def __init__(self, session: AsyncSession, user_id: UUID) -> None:
        self.session = session
        self.user_id = user_id

    def select(self, model: type[DeclarativeBase]) -> "Select[Any]":
        if not hasattr(model, "user_id"):
            raise TypeError(
                f"{model.__name__} has no user_id column — it isn't a user-owned table, so "
                "scoped_session() is the wrong tool here. Use auth_session() (pre-identity "
                "lookups only) or unscoped_session() (health checks etc.) instead."
            )
        return select(model).where(model.user_id == self.user_id)


@asynccontextmanager
async def scoped_session(user_id: UUID) -> AsyncIterator[Scope]:
    """A `Scope` for a known, authenticated user. See module docstring."""
    async with _sessions()() as session:
        async with session.begin():
            yield Scope(session, user_id)