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

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

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

Начну с вопроса: «что такое этот ваш суперапп?» и уже дальше опишу принцип работы и что в итоге получилось воссоздать за 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. Это бета‑версия, поэтому я готов к любой конструктивной критике в свой адрес.