Три архитектурных решения для multi-tenant B2B SaaS, о которых я пожалел, что не узнал раньше

Самая дорогая ошибка моего 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.pyfrom sqlalchemy import selectfrom sqlalchemy.sql import Selectdef 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 upgradeCREATE 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:

``pythonfrom psycopg.errors import ExclusionViolationtry:    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.

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

```pythonimport hashlibimport secretsfrom argon2 import PasswordHasherfrom argon2 import exceptions as argon2_exceptionsPASSWORD_ITERATIONS = 120_000  # legacy PBKDF2 для backward-compatPASSWORD_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 Falsedef 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` после успешной верификации:```pythonif 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 по строкам:

```pythonclass 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:

```bashmake postgres-upDATABASE_URL=... alembic upgrade headDATABASE_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 маскируется существующими объектами.