Самая дорогая ошибка моего B2B SaaS имела ровно одну строчку

```python
  # app/config.py
  TENANT_ID = "tenant-1"
  ```

Когда у меня был один тенант, всё работало корректно. На втором — половина админ-сущностей (врачи, услуги, прайс-листы) начала пропадать из интерфейса клиента. Не «не сохраняться» — а появляться в БД с чужим tenant_id. Я полтора дня смотрел на эту мистику, прежде чем понял: 30 endpoint’ов берут tenant_id из closure из config, а не из user.tenant_id. Очевидно в ретроспективе. Совершенно невидимо во время первого пилота.

Этот разбор — про три архитектурных решения для multi-tenant SaaS в регулируемой отрасли, которые принимаются в первую неделю и потом годами либо экономят месяцы рефакторинга, либо тихо копят техдолг. Стек: Python 3.11 / FastAPI / SQLAlchemy 2.x / PostgreSQL 16. Контекст: B2B SaaS для частных медицинских клиник, 152-ФЗ, real-time scheduling.

TL;DR

- Multi-tenancy через tenant_id column + helper scoped_select с первого коммита — даже если тенант один. Причина внутри.

- Защита от double-booking через PostgreSQL EXCLUDE USING gist + tsrange — реальные цифры: 12 конкурентных запросов на один слот → 1×200, 11×409, без application-level lock'ов.

- 152-ФЗ — это поля и helpers: data_category enum в audit_logs с автоклассификатором, прозрачная миграция с PBKDF2 на argon2id без forced reset, patient_consents со scoped версионированием.

- Один баг с захардкоженным TENANT_ID показывает, почему все три решения нужно закладывать сразу.

- Operational tooling: read-only integrity check на семь категорий аномалий + schema drift detection через alembic check на чистой PG.

Решение №1: multi-tenancy через tenant_id column

Три классических подхода к multi-tenancy в реляционной БД:

Я выбрал третий вариант. Каждая бизнес-таблица содержит tenant_id TEXT NOT NULL с индексом, все запросы обязаны фильтровать по нему. Цена этой простоты — единственная ошибка в фильтре превращается в IDOR на чувствительные данные между тенантами.

Чтобы это не забывать, в проекте есть тонкий helper:

```python
# app/tenant_scope.py
from sqlalchemy import select
from sqlalchemy.sql import Select

def scoped_select(model, user) -> Select:
    """SELECT, автоматически ограниченный текущим тенантом.

    Падает PermissionError, если у user нет tenant_id — это
    намеренно: лучше 500-ка в логах, чем тихая утечка.
    """
    if not user.tenant_id:
        raise PermissionError("User has no tenant_id; refusing to query")
    return select(model).where(model.tenant_id == user.tenant_id)


def scoped_get(db, model, id_, user):
    return db.scalar(
        select(model).where(model.id == id_, model.tenant_id == user.tenant_id)
    )
```

Принцип — лучше явная ошибка, чем тихая утечка. И в code review правило простое: любой select(BusinessModel) без scoped_select — блокирующее замечание.

Bug story в callout: одна строка, два дня дебага

В первой версии я использовал closure tenant_id из config'а во всех write-handler'ах:

```python
> @router.post("/v1/doctors")
> def create_doctor(payload, db, user):
>     item = Doctor(
>         id=f"doctor-{uuid4()}",
>         tenant_id=tenant_id,           # ← closure из config!
>         **payload.model_dump(),
>     )
> ```

Read-path использовал scoped_select(Doctor, user) — корректно. Write-path баковал config.TENANT_ID. Пока тенант один — рассинхрон невидим. Когда подключили вторую клинику, её администратор видел в списке врачей пусто, хотя только что добавил трёх. Эти трое лежали в БД с tenant_id="tenant-1". Фикс: tenant_id=user.tenant_id и регрессия в E2E «зарегистрировать тенанта B → создать сущность как user-B → залогиниться как user-A → убедиться, что не виден».

Урок: helper нужен с первого коммита, даже если тенант один. На десятом тенанте поздно — придётся ревизовать каждый endpoint.

А почему не PostgreSQL Row-Level Security? Это первый вопрос, который мне задают, и он справедливый. RLS дал бы tenant isolation на уровне БД, а не приложения — теоретически надёжнее. На практике RLS требует SET app.tenant_id = '...' в начале каждой транзакции, ломает stateless connection pooling (PgBouncer в transaction mode становится сложнее настраивать), и плохо работает с migrations и admin-задачами, где хочется обойти ограничение легально. Для solo-проекта на ранней стадии overhead RLS не оправдан — scoped_select + lint-правило в code review даёт 95% защиты при 5% сложности. На поздней стадии (десятки enterprise-тенантов с compliance-требованиями) RLS становится разумным дополнением, не заменой.

Решение №2: PostgreSQL EXCLUDE USING gist против double-booking

Два администратора одновременно бронируют один и тот же слот у одного врача. Оба видят слот свободным. Оба вызывают POST /appointments. Оба получают 200 OK. Двойная запись.

Защита «в коде» (SELECT для проверки → INSERT) под параллельной нагрузкой не работает. Между двумя запросами в одной транзакции другая транзакция успевает вставить пересекающийся слот. Я это увидел на первом же load-тесте.

Эволюция защиты на проекте шла в три стадии. Привожу реальные цифры из benchmark'ов на PostgreSQL 16, single-node:

Стадия 1 — application-level check. 12 конкурентных запросов на один слот → 4×200, 8×409. Видимы двойные записи. create_unique_appointments p95 = 26ms.

Стадия 2 — transactional advisory locks (pg_advisory_xact_lock по (doctor_id, time_window) перед INSERT'ом). 12 конкурентных → 1×200, 11×409. Race-condition закрыт, но create_unique_appointments p95 поднялся до 36.86ms из-за serialization overhead.

Стадия 3 — hard DB invariant через EXCLUDE USING gist. Те же 12 конкурентных → 1×200, 11×409, create_unique_appointments p95 = 31.69ms, conflict-path p95 = 11.22ms. Защита переместилась с приложения в schema, advisory lock больше не нужен.

DDL constraint:

``sql
-- Alembic upgrade
CREATE EXTENSION IF NOT EXISTS btree_gist;

ALTER TABLE appointments ADD CONSTRAINT appointments_doctor_slot_excl
  EXCLUDE USING gist (
    doctor_id WITH =,
    tsrange(starts_at, ends_at, '[)') WITH &&
  )
  WHERE (status NOT IN ('cancelled', 'no_show'));
```

Что здесь происходит:

- doctor_id WITH = — два диапазона рассматриваются как конфликтующие, только если у них одинаковый doctor_id.

- tsrange(starts_at, ends_at, '[)') — полуоткрытый интервал; запись 10:00–10:30 и 10:30–11:00 не пересекаются.

- && — оператор пересечения диапазонов.

- WHERE (status NOT IN ('cancelled', 'no_show')) — отменённые и неявки исключены, пациент после отмены может перезаписаться в тот же слот.

btree_gist нужен, потому что = для скалярного doctor_id через стандартный gist-оператор не работает; расширение добавляет btree-семантику внутрь gist-индекса.

Маппинг ошибки в HTTP в FastAPI:

``python
from psycopg.errors import ExclusionViolation

try:
    db.add(appointment)
    db.flush()
except IntegrityError as exc:
    if isinstance(exc.orig, ExclusionViolation):
        raise HTTPException(
            status_code=409,
            detail={
                "code": "APPOINTMENT_SLOT_CONFLICT",
                "message": "Слот занят другой записью",
            },
        )
    raise
```

tsrange или tstzrange? Я выбрал tsrange (naive timestamp без timezone), потому что в этом проекте все timestamp'ы хранятся в UTC, а конверсия в локальное время клиники делается на app-уровне через tenant.timezone. tstzrange работал бы, но добавил бы лишний слой неявных конверсий и потенциальных багов с DST. Если у вас политика хранения «timestamp with timezone везде» — берите tstzrange, разница только в том, как Postgres внутренне трактует диапазоны.

Подводные камни:

- EXCLUDE создаёт gist-индекс, который растёт быстрее B-tree на больших объёмах. По [статье на Хабре про btree_gist benchmark](https://habr.com/ru/articles/820455/) — деградация записи до 2× и чтения до 20% на горячих таблицах. Для таблицы записей пациентов это окей (кардинальность невысокая), для таблицы событий с миллионами строк — задумайтесь.

- WHERE-клауза partial constraint работает только при INSERT/DELETE. Cancellation должна быть UPDATE status='cancelled', не DELETE — иначе запись «исчезает», и слот через секунду занимает кто-то ещё, а отменённую запись уже не восстановить.

- В тестах нужен тот же диалект (PG, не SQLite). SQLite принимает DDL без ошибок, но никаких exclusion constraint не создаёт — баг в тестах не отловить.

Что я получил в итоге: при росте c 1 до 10 клиник в проде ни одной двойной записи. Не нужен Redis, не нужен distributed lock, не нужно application-level retry. Pos­tgreSQL делает всю работу.

Решение №3: 152-ФЗ как код

Для медицинских данных в РФ 152-ФЗ требует: хранения в российских ЦОДах, защиты данных при передаче и хранении, журнала доступа к ПДн, отдельного согласия на обработку специальных категорий, возможности удаления/анонимизации по запросу. Что из этого превращается в код:

Шифрование паролей: миграция PBKDF2 → argon2id без forced reset

Изначально пароли хешились через PBKDF2-HMAC-SHA256 со 120 000 итераций — стандарт OWASP до 2023 года. Современная рекомендация OWASP/NIST — argon2id (memory-hard, устойчивее к GPU-атакам). Резкая миграция через forced password reset недопустима в B2B (десятки сотрудников клиники, регистратор не сможет залогиниться утром понедельника), поэтому работает прозрачная схема: новые пароли пишутся как argon2id, старые верифицируются по своему формату, при успешном логине молча перехешируются в argon2id.

Ключ — детектирование формата по префиксу хеша:

```python
import hashlib
import secrets
from argon2 import PasswordHasher
from argon2 import exceptions as argon2_exceptions

PASSWORD_ITERATIONS = 120_000  # legacy PBKDF2 для backward-compat
PASSWORD_PREFIX = "pbkdf2_sha256"
ARGON2_PREFIX = "$argon2"
PASSWORD_HASHER = PasswordHasher()  # argon2id с дефолтными параметрами


def hash_password(password: str) -> str:
    """Все новые пароли — argon2id."""
    return PASSWORD_HASHER.hash(password)


def verify_password(password: str, encoded: str) -> bool:
    if encoded.startswith(ARGON2_PREFIX):
        try:
            return PASSWORD_HASHER.verify(encoded, password)
        except (argon2_exceptions.VerifyMismatchError,
                argon2_exceptions.InvalidHashError):
            return False
    if encoded.startswith(f"{PASSWORD_PREFIX}$"):
        _, iterations, salt, digest = encoded.split("$", 3)
        computed = hashlib.pbkdf2_hmac(
            "sha256", password.encode(), salt.encode(), int(iterations)
        )
        return secrets.compare_digest(computed.hex(), digest)
    return False


def password_needs_rehash(encoded: str) -> bool:
    if encoded.startswith(f"{PASSWORD_PREFIX}$"):
        return True  # любой PBKDF2 → upgrade в argon2id
    if encoded.startswith(ARGON2_PREFIX):
        return PASSWORD_HASHER.check_needs_rehash(encoded)
    return False
```

В `/auth/login` после успешной верификации:

```python
if password_needs_rehash(user.password_hash):
    user.password_hash = hash_password(plain_password)
    db.commit()
```

Параметры argon2id и trade-off.** Я использую дефолты argon2-cffi: memory_cost=65536 (64 MiB), time_cost=3, parallelism=4. На моём VPS это даёт login latency около 80ms — приемлемо для пользователя, дорого для атакующего (брутфорс на GPU становится непрактичным из-за memory-bound nature алгоритма). Если пилотных клиник будет под 100 одновременно логиниться, можно снизить memory_cost до 32 MiB, но пока я в это ограничение даже близко не упёрся.

Почему argon2id, а не bcrypt или scrypt? bcrypt стабилен, но не memory-hard и проигрывает GPU-фермам. scrypt memory-hard, но менее tunable между tradeoff'ами latency/memory. argon2id — победитель Password Hashing Competition 2015, рекомендация OWASP с 2023, и его подход «hybrid» комбинирует устойчивость argon2i к side-channel атакам с GPU-resistance argon2d. Для регулируемой B2B это разумный default.

Audit log с классификацией категорий ПДн

Каждая операция с данными пациента пишется в audit_logs с полем data_category. Этот enum нужен, чтобы при запросе субъекта (или Роскомнадзора) можно было одним SELECT'ом выгрузить только релевантные категории, а не парсить entity_type/action по строкам:

```python
class AuditLog(Base):
    __tablename__ = "audit_logs"

    id: Mapped[str] = mapped_column(String, primary_key=True)
    tenant_id: Mapped[str] = mapped_column(String, index=True)
    actor_user_id: Mapped[Optional[str]] = mapped_column(String, nullable=True)
    entity_type: Mapped[str] = mapped_column(String)
    entity_id: Mapped[str] = mapped_column(String)
    action: Mapped[str] = mapped_column(String)
    # 152-ФЗ ст. 10 п. 2 — отдельная маркировка спец.категорий ПДн.
    data_category: Mapped[str] = mapped_column(String, default="general", index=True)
    before_json: Mapped[Optional[dict]] = mapped_column(JSON, nullable=True)
    after_json: Mapped[Optional[dict]] = mapped_column(JSON, nullable=True)
    ip: Mapped[Optional[str]] = mapped_column(String, nullable=True)
    created_at: Mapped[datetime] = mapped_column(DateTime, default=utcnow)

    __table_args__ = (
        Index("ix_audit_logs_tenant_created", "tenant_id", text("created_at DESC")),
    )
```

`data_category` — энум на три значения, который проставляется не вручную, а автоматическим классификатором по `entity_type`:

```python
_AUDIT_SPECIAL_HEALTH_ENTITIES = frozenset({
    "patient", "patient_consent",
    "appointment", "encounter", "notification",
    "compliance_retention",
})
_AUDIT_IDENTITY_ENTITIES = frozenset({"user", "user_group"})


def _classify_audit_entity(entity_type: str) -> str:
    if entity_type in _AUDIT_SPECIAL_HEALTH_ENTITIES:
        return "special_health"
    if entity_type in _AUDIT_IDENTITY_ENTITIES:
        return "identity"
    return "general"
```

special_health включает не только диагнозы — туда попадают appointment и notification`, потому что сам факт записи к конкретному врачу + причина обращения уже считаются специальной категорией ПДн в строгой трактовке 152-ФЗ ст. 10. Conservative-классификатор лучше, чем недоклассификация: если регулятор спросит «какие данные о здоровье вы собираете» — лучше ошибиться в сторону «больше», чем «меньше».

Чем 152-ФЗ отличается от GDPR. Один в один шаблон не переносится: 152-ФЗ ближе к американскому HIPAA в том, что отдельно выделяет специальные категории с дополнительными требованиями к согласию и хранению, плюс residency-требование (только РФ-ЦОДы) и обязательная регистрация в реестре операторов ПДн в РКН. GDPR-фреймворк controller/processor/lawful-basis по форме похож, но регулятор и бумажки другие.

Operational tooling

Два инструмента, без которых я бы не доверял БД пилотного проекта.

1. Read-only integrity check. Скрипт на семь стратегий, каждая ловит класс аномалий, который SQL-constraint не отлавливает:

Запускается против локального restore prod-дампа (не против самого prod — long-running scan на горячих таблицах блокирует writers). Exit code 1 при наличии findings — годится как CI-gate перед мажорной миграцией.

2. Schema drift detection.** Перед каждым релизом — full cycle на чистой PG:

```bash
make postgres-up
DATABASE_URL=... alembic upgrade head
DATABASE_URL=... alembic check  # автоген diff против моделей
```

alembic check не идеален: не видит JSONB→JSON рассинхрон между Postgres и SQLite, не различает Index(unique=True) vs UniqueConstraint. Но базовые drift'ы между models.py и реальной схемой ловит надёжно. На локальной dev-БД с накопленным мусором этот workflow не работает — drift маскируется существующими объектами.

Пять выводов

1. Multi-tenant с первого коммита через scoped_select helper и user.tenant_id — даже если тенант один. Захардкоженный config.TENANT_ID — это техдолг, который выстреливает на втором тенанте и требует ревизии каждого endpoint'a.

2. Защиту от race-condition бронирований делать DB-level через EXCLUDE USING gist + tsrange — дешевле и надёжнее application-level locking. Реальные цифры: 12 конкурентных запросов → 1×200, 11×409, p95 31.69ms. Не требует Redis или distributed координации.

3. 152-ФЗ-compliance — это поля и helpers, не маркетинговая галочка. data_category enum в audit_logs с автоклассификатором по entity_type, прозрачная миграция PBKDF2 → argon2id через password_needs_rehash, отдельный patient_consents со scoped версионированием.

4. DB integrity check как операционный артефакт: семь категорий аномалий, которые SQL constraints не ловят. Запуск против локального restore прод-дампа перед мажорным релизом — недорогая страховка.

5. Schema drift detection в CI — alembic upgrade head + alembic check на чистой PG. На dev-БД с накопленным мусором не работает: drift маскируется существующими объектами.