Як побудувати B2B SaaS у Telegram на 1000+ клієнтів і не з’їхати з глузду

💡 Усі статті, обговорення, новини про Python — в одному місці. Приєднуйтесь до Python спільноти!

Всім привіт!

Мене звати Андрій, я Full-Stack Product Engineer та System Architect. Раніше я працював у корпоративному банкінгу (еквайринг), де відповідав за життєвий цикл транзакцій, а зараз спеціалізуюся на розробці безпечних B2B SaaS-платформ, які не гублять гроші і не падають під навантаженням.

Матеріал буде найбільш корисним для Middle+/Senior Backend інженерів, Tech Leads та Software Architects. Якщо ви проєктуєте multi-tenant системи, шукаєте апаратні способи ізоляції даних на рівні ядра БД (PostgreSQL RLS) або хочете перевести свої Telegram-проєкти з «просто скрипта» на сувору Headless-архітектуру (Strict Service Layer) — цей досвід збереже вам десятки годин дебагу та рефакторингу.

Про що саме цей лонгрід

Більшість статей про multi-tenancy на DOU/Habr закінчуються на рівні «додайте tenant_id у кожну таблицю і не забудьте WHERE». Це не архітектура — це міна сповільненої дії, яка вибухне на 50-му клієнті, коли junior забуде WHERE tenant_id = ? у звітному запиті і зіллє дані одного клієнта іншому.

Але... це не туторіал. Це розбір рішень, які я приймав під час проєктування SmartHire — B2B SaaS-платформи у Telegram. Зараз проєкт знаходиться на фінальній стадії перед комерційним запуском: я успішно валідував архітектуру на 1000+ ізольованих тенантів під час стрес-тестів. У цій статті я поділюся тим, які патерни дозволяють закласти Enterprise-надійність ще на стадії розробки, а від яких «класичних» ідей довелося відмовитися через їхню нежиттєздатність.

Що таке SmartHire і чому це не «ще один бот»

Коротко, щоб зафіксувати масштаб і не повертатись до цього питання. SmartHire — це B2B SaaS-платформа з Telegram як основним каналом доставки (поки що). Під капотом чотири окремі фронтенди (Dashboard на React 19 для суперадміна, Mini App для клієнтських тенантів, Onboarding-лендинг, Helm-конфігуратор), 19 FastAPI-роутерів, сім бізнес-модулів через Module Federation (анкети, бронювання, магазин, білінг, рекрутинг, підтримка), білінг з grace period, щоденна auto-reconciliation з банком, OpenTelemetry, Helm chart з HPA на Kubernetes.

Стек без сюрпризів: Python 3.12, aiogram 3.26, FastAPI 0.135, SQLAlchemy 2.0 async, PostgreSQL 14+, Redis 7 з Sentinel, React 19. Жодних екзотичних баз даних, жодних «я написав свій ORM». Усі рішення — стандартні інструменти, застосовані з розумінням trade-off’ів.

Уточнення для тих, хто прочитав «Telegram-бот» і одразу подумав про монорепу з одним bot.py на чотири тисячі рядків і if user_id in admins: ні, це не воно. Той підхід ламається на другому клієнті. Про це — далі.

Multi-tenancy: три школи і чому жодна не підходить у чистому вигляді

Коли інженер вперше стикається з multi-tenancy, у нього є три класичних варіанти, і всі три у вакуумі — погані.

Варіант перший: database-per-tenant. Окремий Postgres instance під кожного клієнта. Isolation ідеальний, регуляторна вимога HIPAA закривається з коробки. Ціна — десятки доларів на місяць за RDS instance, плюс окремий pool з’єднань, окремі міграції, окремий моніторинг. На 1000 тенантах це інфраструктура за $30k/міс і DevOps, який звільниться через три місяці. Працює для enterprise з кількома десятками клієнтів — не для масового SaaS.

Варіант другий: schema-per-tenant. Головний недолік цього підходу — connection pool множиться на кількість schema, а міграції перетворюються на пекло. Для 1000 клієнтів кожну міграцію доведеться проганяти 1000 разів поспіль. Міграція, яка на dev-базі займає 3 секунди, на production може заблокувати деплой на півгодини і призвести до простою (downtime).

Варіант третій: shared schema + колонка tenant_id. Дешево, швидко, масштабовано. І саме той варіант, який junior реалізує найкраще-найгірше: додає колонку, а далі сподівається, що всі розробники не забудуть WHERE tenant_id = ? у кожному запиті. Перший раз, коли хтось забуде — у вас cross-tenant data leak. Питання не «чи трапиться», а «коли і скільки коштуватиме».

У SmartHire я обрав третій варіант, але з трьома шарами захисту, з яких жоден на власноруч не достатній:

Логіка проста: middleware можна обійти через cron job, ORM можна обійти через raw SQL, RLS обійти не можна — це рівень БД. Усі три шари разом дають defense in depth.

Декоратор @tenant_scoped читає tenant_id з contextvars.ContextVar:

def get_tenant_id() -> str:
    """
    Get current tenant_id.
    """
    from core.registry import get_tenant_identity_resolver
    tenant_id = _tenant_id_ctx.get()  # Читає з contextvar
    if not tenant_id:
        return get_tenant_identity_resolver().get_default_tenant_id()
    return tenant_id

А ось як BaseGateway автоматично пробрасує tenant_id у repository — перевіряє при ініціалізації та додає до параметрів запиту:

def __init__(self, pool: Any, tenant_id: str) -> None:
    if not tenant_id or not tenant_id.strip():
        msg = "tenant_id is required for tenant-scoped repositories"
        raise ValueError(msg)
    self.pool = pool
    self.tenant_id = tenant_id.strip()
def _tenant_params(self, params: tuple = ()) -> tuple:
    return (*params, self.tenant_id)  # Автоматично додає tenant_id

PostgreSQL Row-Level Security: як перестати боятись забутого WHERE

RLS у Postgres існує з версії 9.5 — це 2016 рік. За десятиліття вона все ще для більшості розробників «екзотична фіча».

-- Включаємо Row-Level Security
ALTER TABLE ankety ENABLE ROW LEVEL SECURITY;
ALTER TABLE ankety FORCE ROW LEVEL SECURITY;
-- Політика ізоляції: рядки доступні тільки для поточного tenant
CREATE POLICY tenant_isolation_ankety ON ankety
FOR ALL
USING (
    tenant_id = current_setting('app.tenant_id', true)
    OR current_setting('app.is_admin', true)::bool = true
)
WITH CHECK (
    tenant_id = current_setting('app.tenant_id', true)
    OR current_setting('app.is_admin', true)::bool = true
);

Примітка: Друга умова OR current_setting('app.is_admin', true) дозволяє системним операціям (міграції, адмін-панель) обходити RLS.

Аргументів проти три, і всі три не витримують перевірки.

«Це повільно».

Бенчмарк на нашій таблиці ankety з 8.4M рядків: запит SELECT ... WHERE created_at > now() - interval '7 days' без RLS — 47ms p95. З RLS і composite-індексом (tenant_id, created_at) — 51ms p95. Overhead менше 10%, і він ховається в тіні мережевого latency. Якщо у вас RLS дає 2x — ви забули індекс на tenant_id.

«Це складно дебажити».

Так, перші два тижні незрозуміло, чому запит повертає 0 рядків замість очікуваних 100. Потім вчишся одразу перевіряти SELECT current_setting('app.tenant_id') і питання знімаються. Це освоюється швидше за async/await.

«Ми ж і так у коді перевіряємо (через ORM чи Middleware)».

Це аргумент, який працює лише на стадії пет-проєкту. У реальному SaaS рано чи пізно хтось напише raw SQL для складної аналітики, фонової джоби чи міграції. І цей «хтось» (ви самі о 2-й ночі, новий розробник у команді або навіть AI-асистент) забуде додати WHERE tenant_id = ?.

Покладатися виключно на дисципліну розробників у питаннях ізоляції даних — це антипатерн. PostgreSQL RLS діє як страховка від людського фактору на найнижчому рівні. Вона гарантує: якщо помилка в application-шарі все ж трапиться, база даних просто поверне 0 rows, замість того щоб влаштувати крос-тенантний витік даних, який закінчиться судовим позовом.

Технічно це працює так. У middleware на кожен запит робимо SET LOCAL app.tenant_id = :id всередині транзакції. Ключове слово «LOCAL», бо без нього налаштування переживе commit і прилипне до connection у pool, а наступний запит від іншого тенанта виконається з чужим tenant_id. Це шлях до катастрофи.

FastAPI Middleware + Transaction + SET LOCAL

Мідлвар отримує tenant_id і зберігає в contextvars. При отриманні з’єднання з пулу виконується SET LOCAL app.tenant_id:

@app.middleware("http")
async def tenant_context_middleware(request: Request, call_next: Any) -> Any:
    """Extract tenant_id and set in contextvars."""
    tenant_id = getattr(request.state, "tenant_id", "default")
    set_tenant_id(tenant_id)  # Зберігає в ContextVar
    if hasattr(app.state, "di_container") and app.state.di_container:
        request.state.container = app.state.di_container
    response = await call_next(request)
    inject_traceparent_header(response.headers)
    return response

А ось як пул встановлює app.tenant_id на з’єднанні (аналог SET LOCAL):

async def _setup_connection_context(self, conn: Any) -> tuple[bool, str | None]:
    """Set up RLS or tenant_id on the connection."""
    # ... bypass check ...
    
    tenant_id = None
    with contextlib.suppress(Exception):
        from core.tenant_context import get_tenant_id
        tenant_id = get_tenant_id()  # Читає з ContextVar
    if tenant_id and tenant_id != "default":
        # SET LOCAL app.tenant_id = :id (діє до кінця сесії/транзакції)
        await conn.execute("SELECT set_config('app.tenant_id', $1, true)", tenant_id)
    return False, tenant_id

Важливий нюанс: якщо ви налаштовуєте RLS на рівні checkout’у з пулу з’єднань (до старту транзакції), використовуйте set_config(..., false) і обов’язково робіть RESET app.tenant_id у блоці finally при поверненні з’єднання в пул. Інакше tenant_id «отруїть» з’єднання для наступного клієнта. Якщо ж ви обгортаєте запит у async with conn.transaction():, безпечніше використовувати set_config(..., true) — тоді Postgres сам очистить контекст після коміту/ролбеку.

Окремий кейс — це аналітика. Іноді треба зробити cross-tenant звіт («скільки всього грошей пройшло через платформу за квартал»). Для цього я створив окрему роль analytics_readonly з атрибутом BYPASSRLS — вона єдина може ігнорувати policy. Кожен запит з-під цієї ролі логується в окремий audit log. Якщо хтось колись витягне з неї дані без права — це можна буде побачити за хвилину, а не за квартал.

Headless Service Layer: чому хендлер не повинен знати про БД

Якщо ваш handler aiogram виглядає так — далі не читайте, спочатку зробіть рефакторинг:

@router.message(F.text == "/мої анкети")
async def my_ankety(message: Message, session: AsyncSession):
    result = await session.execute(
        select(Anketa).where(Anketa.user_id == message.from_user.id)
    )
    ankety = result.scalars().all()
    text = "\n".join(f"{a.id}: {a.title}" for a in ankety)
    await message.answer(text)

Тут все зламано одразу: handler знає про БД, бізнес-логіка змішана з форматуванням, відсутній tenant_id, цей же кейс неможливо викликати з REST API без копіпасти. Спробуйте написати на цей handler unit-тест без mock’а Telegram. Не вийде і це не ваша проблема, це проблема архітектури.

У SmartHire діє «Service Layer Enforcement». Правило просте: handlerам (і aiogram, і FastAPI) забороняється імпортувати з core/database/ або core/repositories/ напряму. Перевірка — кастомний ruff-rule, який падає на pre-commit через Lefthook. Якщо ви якось примудрилися закомітити — CI вб’є PR на стадії lint.

Шари виглядають так:

handlers/ ← transport (aiogram, FastAPI)
 ↓
core/services/ ← domain logic (headless, не знає про transport)
 ↓
core/repositories/ ← data access (tenant-scoped)
 ↓
core/database/ ← SQLAlchemy models, raw queries

Service приймає tenant_id як обов’язковий аргумент — не читає його з якогось current_user, не лізе в contextvar напряму. Це робить service викликаним з будь-якого entrypoint: бот, API, scheduler, тест, manage-команда. Той самий AnketaService.create() працює і з Telegram-команди, і з REST endpoint, і з cron’а, який імпортує заявки з зовнішнього джерела.

Порівняння: Legacy vs SmartHire стилі

Legacy (Handler з SQL):

async def _handle_new_anketa(msg: Message) -> None:
    """Створення анкети — SQL у хендлері."""
    tenant_id = get_tenant_id()
    try:
        pool = get_container().get_db_pool()
        if pool:
            async with pool.acquire() as conn:
                anketa_id = generate_anketa_id()
                await conn.execute(
                    """INSERT INTO ankety
                         (anketa_id, desc, price, phone, video_id, 
                        message_id, status, tenant_id, created_at)
                       VALUES ($1, $2, $3, $4, $5, $6, 'opened', $7, NOW())""",
                    anketa_id, msg.text, price, phone, video_id, 
                    msg.message_id, tenant_id
                )
    except asyncpg.PostgresConnectionError as e:
        logger.error("Insert failed: %s", e)

Проблеми: SQL у хендлері, немає ізоляції бізнес-логіки, важко тестувати, дублювання коду.

SmartHire (Handler → Service → Repository)

# core/services/anketa_service.py:161-190
async def save_anketa(
    self,
    anketa_id: str,
    desc: str,
    price: int,
    phone: str,  # ← raw phone на вході
    video_id: str,
    message_id: int,
    hashtag: str,
    candidate_name: str | None = None,
) -> None:
    """
    Save a new anketa to DB.
    
    Args:
        phone: Candidate phone (raw, will be encrypted)
    """
    from utils.crypto import encrypt_phone
    # Шифрування — бізнес-логіка, в Service Layer
    encrypted_phone = await asyncio.to_thread(encrypt_phone, str(phone))
    
    # Repository отримує вже зашифровані дані
    await self.anketa_gateway.save_anketa(
        anketa_id, desc, price, encrypted_phone, video_id, 
        int(message_id), hashtag, candidate_name or ""
    )

Переваги: чистий поділ шарів, легко тестувати (мок repository), перевикористання логіки.

AnketaService.save_anketa()

# core/repositories/anketa.py:326-337
async def save_anketa(
    self,
    anketa_id: str,
    desc: str,
    price: int,
    encrypted_phone: str,  # ← вже зашифрований
    video_id: str,
    message_id: int,
    hashtag: str = "",
    candidate_name: str = "",
) -> None:
    """Upsert (create or update) an anketa record with locking."""
    # Тільки SQL — жодної криптографії
    query = """INSERT INTO vacancies ..."""
    await self._execute(query, ...)

Ключовий момент: Жодного імпорту з aiogram чи FastAPI у сервісі. Репозиторій — «тупий» (тільки SQL), сервіс — бізнес-логіка (шифрування PII).

Unit-тест без моку Telegram

# tests/unit/core/services/test_anketa_service.py
@pytest.mark.asyncio
async def test_save_anketa_persists_to_db(gateways) -> None:
    # Arrange — реальні gateway з in-memory DB
    anketa_gw, _ = gateways
    from core.repositories.business_config import BusinessConfigGateway
    from core.services.anketa_service import AnketaService
    service = AnketaService(
        anketa_gateway=anketa_gw,
        business_config_gateway=BusinessConfigGateway(
            pool=anketa_gw.pool, tenant_id="test"
        ),
    )
    # Act — виклик сервісу без Telegram
    await service.save_anketa(
        anketa_id="A123",
        desc="Python Dev",
        price=50000,
        phone="+380501234567",
        video_id="video123",
        message_id=999,
        hashtag="python",
    )
    # Assert — перевірка в БД
    result = await anketa_gw.get_anketa("A123")
    assert result is not None
    assert result["desc"] == "Python Dev"
    assert result["status"] == "opened"

Доказ headless: Тест працює без жодного моку Telegram — використовує реальну in-memory SQLite базу та реальні gateway. Це підтверджує, що бізнес-логіка повністю відокремлена від інфраструктури.

DI-контейнер на dependency-injector тримає це разом. Жодних from core.services import ankety_service на module-level — все через Depends() у FastAPI або через явний інжект у aiogram через middleware. Це здається бюрократією на старті, але рятує життя при першому ж великому рефакторингу.

Module Federation: як додати новий бізнес-напрямок за 2 дні

Класична траєкторія SaaS, який не передбачив modularity на старті: клієнт А просить feature X — додаєте if client_id == "A". Клієнт B хоче Y — ще один if. Клієнт C хоче поведінку як у А, але з нюансом — if client_id == "A" or client_id == "C". Через рік у вас 200 умовних гілок, кожен реліз ламає двох клієнтів, онбординг нового тенанта = два тижні бекенд-роботи.

У SmartHire є BaseModule — абстрактний клас з контрактом на чотири методи: setup(dp, app, container), get_catalog_items(), get_routes(), get_health(). Все, що потрібно зробити для нового бізнес-напрямку — успадкувати, реалізувати ці методи, скласти у core/modules/. ModuleRegistry підхопить автоматично при старті.

Зараз у нас сім модулів: Catalog (анкети — той самий рекрутинговий core), Booking (бронювання слотів), Shop (товари і замовлення), Billing (підписки і платежі), Recruiting (воронка кандидатів), Support (тікет-система), плюс ще один варіант Shop під специфіку конкретної ніші. Жоден з модулів не імпортує інших — спілкуються через event bus на Redis pub/sub.

Per-tenant вмикання — через DB-config і feature flags. Тенант А включає [catalog, billing], тенант B — [booking, shop, support]. Перемикач у Dashboard, миттєвий, без релізу.

BaseModule (ABC)

class BaseModule(ABC):
    """
    Contract for all pluggable business modules.
    Subclasses declare their module_id, feature flag linkage and menu buttons.
    The registry uses is_active() to determine whether the module is enabled
    for a given tenant.
    """
    feature_flag: FeatureFlag | None = None
    is_enabled_by_default: bool = False
    @property
    @abstractmethod
    def module_id(self) -> str:
        """Unique snake_case identifier, e.g. 'recruiting'."""
    @property
    @abstractmethod
    def display_name(self) -> str:
        """Human-readable name shown in Dashboard / menu."""
    @property
    @abstractmethod
    def description(self) -> str:
        """Short description for Dashboard module card."""
    @property
    @abstractmethod
    def icon(self) -> str:
        """Emoji or icon identifier for UI."""
    async def is_active(self, tenant_id: str | None = None) -> bool:
        """Check linked feature_flag via FeatureFlagService."""
        if self.feature_flag is None:
            return self.is_enabled_by_default
        return await get_feature_flag_checker().is_enabled(self.feature_flag, tenant_id)

BookingModule

class BookingModule(BaseModule):
    """Service booking and appointment scheduling module."""
    feature_flag: FeatureFlag | None = FeatureFlag.BOOKING_ENABLED
    is_enabled_by_default: bool = False
    @property
    def module_id(self) -> str:
        return "booking"
    @property
    def display_name(self) -> str:
        return "Бронювання"
    @property
    def description(self) -> str:
        return "Запис на послуги та бронювання часу"
    @property
    def icon(self) -> str:
        return "📅"
    async def get_menu_buttons(self, tenant_id: str) -> list[InlineKeyboardButton]:
        return [
            InlineKeyboardButton(
                text="📅 Запис на послуги",
                callback_data=ModuleEntryCb(module_id="booking", action="menu").pack(),
            )
        ]
    def get_onboarding_manifest(self) -> ModuleManifest:
        return ModuleManifest(
            module_id=self.module_id,
            title=self.display_name,
            description=self.description,
            onboarding_fields=[
                OnboardingField(
                    key="service_label",
                    type="text",
                    label="Назва сутності",
                    question="Як називати основну послугу в боті?",
                    placeholder="послуга",
                    default="послуга",
                ),
                OnboardingField(
                    key="currency",
                    type="select",
                    label="Валюта",
                    question="У якій валюті показувати ціни?",
                    default="UAH",
                    options=[
                        OnboardingFieldOption(value="UAH", label="UAH"),
                        OnboardingFieldOption(value="USD", label="USD"),
                    ],
                ),
            ],
        )

Висновок: Всього ~80 рядків коду (включаючи BaseModule) — і готова нова вертикаль бізнесу!

Чому 0.1 + 0.2 != 0.3, або як ваш код краде гроші бізнесу

Модулі, багаторівнева ізоляція та сувора Headless-архітектура — це бетонний фундамент. З таким фундаментом ви можете спати спокійно, знаючи, що дані клієнтів у безпеці, а база витримає будь-який наплив трафіку.

Але навіть найчистіший код не врятує ваш бізнес, якщо він не вміє рахувати гроші.

У наступній (другій) частині статті я розкажу про фінансову інженерію та Observability. Ви дізнаєтесь:

• Чому я на рівні лінтерів заборонив використовувати float для грошей (і чому ваш код зараз, ймовірно, краде копійки).
• Чому webhook’ам від платіжних систем не можна вірити і як я побудував 100% ідемпотентність.
• Які 5 метрик у Grafana дозволяють повністю контролювати здоров’я 1000+ ботів.
• Zero-downtime деплой: як оновлювати систему без втрати жодного HTTP-запиту.

Підписуйтесь на мій LinkedIn, щоб не пропустити другу частину.

А поки що — чекаю вас у коментарях. Розкажіть, які милиці ви використовували для реалізації multi-tenancy і чи спасав вас коли-небудь RLS?

(P.S. Більше архітектурних схем та ADR я виклав у публічний GitHub-репозиторій SmartHire-Showcase).