Как я смог запустить суперапп и что это на самом деле

Всем привет! Пишу ознакомительную статью о моем творении и как работают супераппы, если возникнут вопросы, обращайтесь в коменты.

Первый шаг в бездну

Начну с вопроса: «что такое этот ваш суперапп?» и уже дальше опишу принцип работы и что в итоге получилось воссоздать за 8 месяцев непрерывного кодинга. Суперапп — это объединение двух и более функционалов нескольких сервисов в одно единое приложение или веб‑приложение. Проще говоря, его задача избавить обычного юзера интернета от лишних вкладок на компе и сделать все более централизованным. Хотя тут скорее зависит от того, на какой суперапп мы смотрим.

Архитектура подобных проектов всегда упирается в четыре важные вещи: бэкенд, фронтенд, сервер и защита всего этого. От сюда уже идут другие, не менее важные нюансы, которые следует упомянуть. А что вообще совмещать? Мой ответ, на такой вопрос прост и банален: Маркет и Мессенджер — что я и сделал. Звучит нереально, однако это реально, но крайне трудно, если у вас нет опыта и идеи как все это можно организовать. Хотя пожалуй самый важный аспект это безопасность и защита от IDOR атак. Но и они в этом сценарии отлетают, так как их крайне легко отражать проверками на уровне сервера.

Вот банальный пример, который гарантирует защиту, но при этом является одним из простейших защитных методов. Предположим, у нас есть WebSocketManager с обработкой подключений в consumers.py. Там реализован универсальный базовый метод connect. Для подключения пользователя к беспрерывному соединению с чатом как раз на этом уровне следует внедрять проверку безопасности. Делается это следующим образом:

@database_sync_to_async
    def is_user_in_chat(self, user, chat_id):
        # Проверяем, существует ли чат и входит ли в него данный юзер
        chat = Chat.objects.filter(pk=chat_id, participants=user).exists()
        if chat: 
            return chat
        else: 
            chat = Chat.objects.filter(pk=chat_id, subscribers=user).exists()
            return chat

    # 1. Подключение
    async def connect(self):
        try:
            # Получаем chat_id с проверкой
            self.chat_id = str(self.scope['url_route']['kwargs']['chat_id'])
            self.room_group_name = f"chat_{self.chat_id}"
            self.user = self.scope["user"]

            # (Безопасность соединения!) 
            # Проверка если юзер не аутентифицирован никак в чате: 
            user_has_access = await self.is_user_in_chat(self.user, self.chat_id)
            if not user_has_access:
                logger.warning(f"User {self.user.id} tried to access chat {self.chat_id} without permission!")
                await self.close(code=4403) # Закрываем соединение (Forbidden)
                return
            
            # остальные проверки
            #  ...

Таким образом происходит четкая валидация пользователей в каждом чате, следовательно утечка данных полностью пресекается, а использование проекта становится абсолютно безопасным.

Проект в вебе, поэтому стек у меня банальный для такого масштаба: Django на бэкенде (Python) и JavaScript на фронтенде. Как я уже упоминал, на разработку ушло 8 месяцев, и всё это время задачи были четко распределены по этапам.

Первые два месяца ушли на верстку самого сайта — дизайн, архитектуру фронтенда и базовую функциональность кнопок. В этот срок удалось уложиться идеально. По сути, так же гладко прошли и остальные этапы, кроме интеграции коммерческого API. Тут пришлось провести тотальную работу, чтобы соответствовать стандартам платежных систем и логистических сервисов.

Второй шаг в бездну — погружение

Следующим этапом (с третьего по пятый месяц) стало создание своего API и синхронизация базы данных. Пожалуй, это самое захватывающее и безусловно самое важное. Задача заключалась в построении единой системы защиты. Данные, передаваемые между пользователями, ясное дело, должны как-то шифроваться и быть недоступны посторонним.

Тут обычно есть два пути: либо использовать сквозное шифрование (E2EE), либо защищать данные в движении с помощью TLS‑протоколов, а на самом сервере шифровать их при хранении в базе данных с использованием секретных ключей. Выбор почти у всех всегда один — второй.

Это объясняется тем, что так проще управлять данными и реализовывать привычный бэкенд-функционал. Ведь в случае с E2EE данные становятся абсолютно «слепыми» для сервера. Мне кажется, что сквозной шифр эффективно подошел бы только для классического, изолированного чата без каких-либо надстроек и сложной бизнес-логики.

В качестве примера второго варианта шифрования (на стороне сервера) можно взять специализированный класс кастомного поля для моделей Django. Он перехватывает текстовые данные при записи в базу и автоматически расшифровывает их при чтении, гарантируя защиту данных «из коробки»:

class EncryptedTextField(models.TextField):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._cipher = None
    
    @property
    def cipher(self):
        if self._cipher is None:
            try:
                from cryptography.fernet import Fernet
                if settings.ENCRYPTION_KEY:
                    key = settings.ENCRYPTION_KEY.encode()
                    self._cipher = Fernet(key)
                else:
                    logger.warning("ENCRYPTION_KEY not set, encryption disabled")
            except Exception as e:
                logger.error(f"Failed to initialize cipher: {e}")
        return self._cipher
    
    def from_db_value(self, value, expression, connection):
        try:
            encrypted_bytes = base64.b64decode(value)
            decrypted_bytes = self.cipher.decrypt(encrypted_bytes)
            return decrypted_bytes.decode('utf-8')
        except (base64.binascii.Error, ValueError):
            return value
        except Exception as e:
            # Ошибка расшифровки
            logger.error(f"Decryption error: {e}")
            return f"[DECRYPT_ERROR: {value[:30]}...]"
    
    def get_prep_value(self, value):
        if value is None:
            return None
        
        # Если значение уже зашифровано или это ошибка
        if isinstance(value, str) and (value.startswith('gAAAAA') or value.startswith('[DECRYPT_ERROR:')):
            return value
        try:
            # Шифруем
            encrypted_bytes = self.cipher.encrypt(value.encode('utf-8'))
            return base64.b64encode(encrypted_bytes).decode('utf-8')
        except Exception as e:
            logger.error(f"Encryption error: {e}")
            return value

Ну и, пожалуй, самое приятное — это логика работы Маркета. Здесь разработка шла по довольно понятному сценарию, который можно описать классическими методами записи и выборки данных из БД. Всё устроено прозрачно: на бэкенде отрабатывают триггеры и проверки, определяющие, будет ли опубликован конкретный товар и соответствует ли он правилам платформы.

Намного интереснее устроены алгоритмы ранжирования, или, проще говоря, лента листингов. Для её формирования нужен полноценный перебор и скоринг данных. Система начисляет определенные баллы за каждый кейс: например, учитывается дата публикации, (триггер для моментального взлета листинга вверх ), количество просмотров, количество лайков, количество заказов на объявление. На основе финального веса товара Django формирует персональную выдачу для каждого пользователя в реальном времени.

Пример такой работы можно представить в виде выделенной функции, которая обрабатывает каждый запрос по листингам и возвращает отсортированный кверисет с использованием встроенных инструментов агрегации Django ORM:

def get_listings(search_query=None):
    base_qs = Listing.objects.filter(
        delete_status=False,
        is_available=True,
        successfull_flag=True,
    ).select_related('category').only(
        'name', 'description', 'type_listing', 'state_listing',
        'address', 'price_listing', 'sum_listing', 'delivery_state',
        'rating', 'model_type', 'created_at', 'category__name'
    )
    
    # Параметры алгоритма
    new_listing_boost_hours = 1
    base_rating_weight = 1.0
    newness_weight = 2.0
    
    def annotate_listings(queryset):
        return queryset.annotate(
            hours_since_creation=ExpressionWrapper(
                ExtractHour(timezone.now() - F('created_at')),
                output_field=FloatField()
            ),
            is_new=Case(
                When(hours_since_creation__lt=new_listing_boost_hours, 
                     then=Value(newness_weight)),
                default=Value(0),
                output_field=FloatField()
            ),
            weighted_score=ExpressionWrapper(
                F('rating') * Value(base_rating_weight) + 
                F('is_new') * (Value(newness_weight) - F('hours_since_creation')/Value(new_listing_boost_hours)),
                output_field=FloatField()
            )
        )
    result = list(annotate_listings(base_qs).order_by('-weighted_score', '-created_at'))
    return result

Третий шаг в бездну — всплытие

После разбора основных технических нюансов нельзя не упомянуть, что я разрабатывал весь проект в соло. Поэтому мои рассуждения и технические придирки сформулированы только на личном опыте. Я всегда буду рад услышать любой комментарий в свой адрес и ознакомиться с мнением, так сказать, народа и читателей данной статьи.

Супераппом уже можно воспользоваться и ознакомиться лично: vendergram.com. Это бета‑версия, поэтому я готов к любой конструктивной критике в свой адрес.