gendesign/tradein-mvp/backend/app/services/identity_store.py

"""Единственное место, знающее, В КАКОЙ БД и В КАКИХ ТАБЛИЦАХ живёт identity.

Эпик «единый вход»: люди «Меры» (trade-in) и «Птицы» (Site Finder) переезжают в
общую БД `auth` (`users` / `sessions`, миграции data/sql/auth/001-004), а
`tradein_users` в итоге удаляется. Переезд идёт под флагом
`settings.identity_store`, дефолт которого = СТАРОЕ поведение:

    "tradein" (ДЕФОЛТ) — БД tradein, tradein_users / tradein_sessions;
    "auth"             — БД auth,    users / sessions.

Смысл модуля: во всём остальном коде не должно быть ни одного упоминания
конкретной БД, конкретных имён таблиц и того, каким столбцом выражено состояние
доступа. Кто хочет читать/писать людей и сессии — спрашивает здесь.

Что модуль отдаёт вызывающему:
  * `identity_session()` / `get_identity_db()` — сессия ТОЙ БД, которая сейчас
    является реестром (для "tradein" это ровно `app.core.db.SessionLocal`, то
    есть сегодняшний прод-путь без единого лишнего коннекта);
  * `identity_schema()` — имена таблиц users/sessions и имя колонки состояния
    доступа;
  * `AccessState` + `to_access_state()` — ОДНО понятие «состояние доступа» для
    обеих схем.

Схемы `tradein_users` и `auth.users` совпадают, кроме состояния доступа:
`tradein_users.is_active` — boolean, `auth.users.access_state` — text из трёх
значений (`active` / `trial_expired` / `disabled`, семантика — в COMMENT'е
миграции 004). Вызывающий код обязан работать с ОДНИМ понятием: он читает
колонку `schema.access_state_column` и прогоняет значение через
`to_access_state()`. Второго представления состояния в коде быть не должно —
`if row.is_active` вне этого модуля больше не пишем.

Как СПРАШИВАТЬ состояние доступа (канонический вызов):

    schema = identity_schema()
    with identity_session() as db:
        row = db.execute(
            text(
                f"SELECT u.id, u.username, u.role, "
                f"       u.{schema.access_state_column} AS access_state "
                f"  FROM {schema.users_table} u "
                f" WHERE u.username = :username"
            ),
            {"username": username},
        ).fetchone()
    state = to_access_state(row.access_state)
    if not state.can_sign_in:
        ...  # 401 для disabled, отдельный 403 для AccessState.TRIAL_EXPIRED

Значение подставляется bind-параметром (`:username`), имя таблицы и имя колонки —
из `schema`, то есть из фиксированного словаря; в SQL-строку не попадает ничего,
пришедшего снаружи.

Как ПИСАТЬ состояние доступа (обратное направление, `access_state_param()`):

    db.execute(
        text(
            f"UPDATE {schema.users_table} "
            f"   SET {schema.access_state_column} = :access_state "
            f" WHERE id = :id"
        ),
        {"access_state": access_state_param(AccessState.DISABLED), "id": user_id},
    )

Литералов `True` / `'active'` по месту быть не должно: тип колонки разный, и
единственное место, знающее какой, — этот модуль.

⚠️ SQL-инъекция по имени таблицы: имена таблиц/колонок в SQL нельзя передать
bind-параметром, поэтому они подставляются в строку запроса. Единственный
допустимый источник — фиксированный словарь `_SCHEMAS` НИЖЕ. Никакой
конкатенации с внешним вводом (заголовок, тело запроса, переменная окружения,
имя роли) — значение `settings.identity_store` ограничено `Literal` в pydantic,
и лукап по нему делается только здесь.
"""

from __future__ import annotations

import logging
from collections.abc import Generator, Iterator
from contextlib import contextmanager
from dataclasses import dataclass
from enum import StrEnum
from typing import Annotated

from fastapi import Depends
from sqlalchemy.orm import Session

from app.core import auth_db
from app.core.config import settings
from app.core.db import SessionLocal, get_db

logger = logging.getLogger(__name__)


class AccessState(StrEnum):
    """Состояние доступа аккаунта — ЕДИНОЕ понятие для обеих схем.

    Значения дословно совпадают с `auth.users.access_state` (CHECK-констрейнт
    `users_access_state_ck`, миграция 004); булев `tradein_users.is_active`
    приводится сюда в `to_access_state()`.

    Семантика (COMMENT миграции 004, решение владельца от 2026-07-31):
      active        — вход разрешён;
      trial_expired — пароль ВЕРНЫЙ, но пробный период истёк: отдельный 403 и
                      экран «пробный доступ закончился», сессия не выдаётся;
      disabled      — доступ закрыт: generic 401, для пользователя неотличимо от
                      неверного пароля.
    Неверный пароль в ЛЮБОМ состоянии → generic 401, иначе отдельный ответ для
    trial_expired превращается в оракул существования логина.
    """

    ACTIVE = "active"
    TRIAL_EXPIRED = "trial_expired"
    DISABLED = "disabled"

    @property
    def can_sign_in(self) -> bool:
        """True только для `active` — единственная проверка «пускать ли».

        Вынесена в свойство, чтобы вызывающий не писал `state == "active"`:
        добавится четвёртое состояние — оно по умолчанию окажется «не пускать»,
        а не «пускать, потому что не disabled».
        """
        return self is AccessState.ACTIVE


@dataclass(frozen=True, slots=True)
class IdentitySchema:
    """Где физически лежит identity при текущем значении флага.

    Attributes:
        store: значение `settings.identity_store`, которому соответствует схема.
        users_table: имя таблицы людей.
        sessions_table: имя таблицы сессий.
        access_state_column: имя колонки состояния доступа. Значение из неё
            ОБЯЗАНО пройти через `to_access_state()` — тип отличается между
            схемами (boolean против text).
        access_state_sql_type: SQL-тип этой колонки для `CAST(:param AS ...)`.
            Нужен там, где параметр может быть NULL (`COALESCE(CAST(:x AS T), col)`
            в PATCH «Команды»): без явного типа Postgres не может вывести тип
            NULL-параметра. Значение — литерал из `_SCHEMAS`, в SQL-строку
            снаружи ничего не попадает.
    """

    store: str
    users_table: str
    sessions_table: str
    access_state_column: str
    access_state_sql_type: str


# Фиксированный словарь — ЕДИНСТВЕННЫЙ источник имён таблиц/колонок для SQL.
# Ключи = допустимые значения settings.identity_store (Literal в pydantic).
_SCHEMAS: dict[str, IdentitySchema] = {
    "tradein": IdentitySchema(
        store="tradein",
        users_table="tradein_users",
        sessions_table="tradein_sessions",
        access_state_column="is_active",
        access_state_sql_type="boolean",
    ),
    "auth": IdentitySchema(
        store="auth",
        # В БД `auth` таблицы лежат без префикса продукта — реестр общий
        # (data/sql/auth/001_identity_schema.sql).
        users_table="users",
        sessions_table="sessions",
        access_state_column="access_state",
        access_state_sql_type="text",
    ),
}


def identity_schema() -> IdentitySchema:
    """Схема реестра для текущего значения `settings.identity_store`.

    Читается на КАЖДОМ вызове, а не кешируется на импорте: тесты и
    переключение флага не должны требовать перезагрузки модулей.
    """
    schema = _SCHEMAS.get(settings.identity_store)
    if schema is None:
        # Недостижимо через настройки (Literal валидируется pydantic), но
        # молчаливый fallback здесь означал бы поход не в ту БД.
        raise ValueError(f"неизвестный identity_store={settings.identity_store!r}")
    return schema


@contextmanager
def identity_session() -> Iterator[Session]:
    """Сессия БД, в которой сейчас живёт identity.

    "tradein" → `app.core.db.SessionLocal` (та же БД и тот же пул, что у всего
    остального приложения — сегодняшнее поведение прода без изменений).
    "auth"    → ленивый engine `app.core.auth_db`; пустой `AUTH_DATABASE_URL`
    здесь поднимет `AuthDatabaseNotConfiguredError`, а не отдаст пустой
    результат.
    """
    if settings.identity_store == "auth":
        with auth_db.auth_session() as db:
            yield db
    else:
        with SessionLocal() as db:
            yield db


def get_identity_db(
    db: Annotated[Session, Depends(get_db)],
) -> Generator[Session, None, None]:
    """FastAPI-зависимость: `db: Annotated[Session, Depends(get_identity_db)]`.

    Аналог `app.core.db.get_db`, но для реестра людей. Роуты, работающие с
    identity, обязаны брать сессию отсюда — иначе при `identity_store="auth"`
    они уйдут запросом в БД tradein, где нужных таблиц уже не будет.

    ⚠️ При `identity_store="tradein"` отдаётся РОВНО ТОТ ЖЕ объект `Session`,
    что и у `Depends(get_db)` — не новая сессия к той же БД. Это не экономия
    коннекта, а требование «прод обязан работать точно как сейчас»: роуты
    «Команды» пишут в ОДНОЙ транзакции строку сотрудника (реестр) и его квоту
    (`account_quota_overrides`, продуктовая таблица). Две сессии = две
    транзакции = состояние «сотрудник создан, квота нет» на ровном месте.
    FastAPI кеширует результат `Depends(get_db)` в пределах запроса, поэтому
    роут, объявивший ОБЕ зависимости, в этом режиме получает один и тот же
    объект, и `db is identity_db` — честный рантайм-признак «одна БД».

    При `identity_store="auth"` это разные БД физически, и одной транзакции
    быть не может (двухфазный коммит здесь не заводим): вызывающий код обязан
    коммитить обе сессии и понимать порядок — см. `app.api.v1.team`.
    Зависимость `get_db` при этом всё равно резолвится, но `Session` ленив —
    без единого запроса он коннект не открывает, так что лишнего соединения с
    БД tradein не появляется.
    """
    if settings.identity_store != "auth":
        yield db
        return
    with auth_db.auth_session() as identity_db:
        yield identity_db


def to_access_state(value: object) -> AccessState:
    """Приводит значение колонки состояния доступа к `AccessState`.

    ЕДИНСТВЕННОЕ место, где булев `tradein_users.is_active` превращается в
    трёхзначное состояние: True → `active`, False → `disabled` (жёсткая
    блокировка, generic 401 — ровно то, что булева схема и означала).
    `trial_expired` в булевой схеме выразить нечем: состояния там не
    существовало, и на tradein-пути оно не появится.

    Fail-closed: неизвестная строка, NULL и любой неожиданный тип → `disabled` +
    WARNING. Обратный выбор (пускать всё, что не `disabled`) означал бы, что
    новое состояние, добавленное миграцией раньше кода, молча раздаёт доступ.
    """
    if isinstance(value, bool):
        return AccessState.ACTIVE if value else AccessState.DISABLED
    if isinstance(value, str):
        try:
            return AccessState(value)
        except ValueError:
            logger.warning(
                "identity_store: неизвестное состояние доступа %r → трактую как disabled", value
            )
            return AccessState.DISABLED
    logger.warning(
        "identity_store: состояние доступа %r неожиданного типа %s → трактую как disabled",
        value,
        type(value).__name__,
    )
    return AccessState.DISABLED


def access_state_param(state: AccessState) -> bool | str:
    """Значение для ЗАПИСИ в `schema.access_state_column` — обратная к `to_access_state()`.

    Тип колонки разный (boolean против text), поэтому конверсию нельзя оставить
    вызывающему: он бы неизбежно писал `True`/`'active'` по месту, и это ровно
    то второе представление состояния, которого в коде быть не должно.

    Для булевой схемы `trial_expired` невыразим — там существуют только «пустят»
    и «не пустят», и попытка записать промежуточное состояние молча стала бы
    жёсткой блокировкой (клиент увидел бы «неверный пароль» вместо экрана
    пробного периода). Поэтому это ошибка вызывающего, а не тихое приведение:
    писать `trial_expired` можно только при `identity_store="auth"`.
    """
    schema = identity_schema()
    if schema.access_state_sql_type == "boolean":
        if state is AccessState.TRIAL_EXPIRED:
            raise ValueError(
                f"состояние {state.value!r} невыразимо в схеме {schema.store!r} "
                f"(колонка {schema.access_state_column} — boolean): доступны только "
                f"{AccessState.ACTIVE.value!r} и {AccessState.DISABLED.value!r}"
            )
        return state.can_sign_in
    return state.value