Всем привет! Пишу ознакомительную статью о моем творении и как работают супераппы, если возникнут вопросы, обращайтесь в коменты.
Первый шаг в бездну
Начну с вопроса: «что такое этот ваш суперапп?» и уже дальше опишу принцип работы и что в итоге получилось воссоздать за 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
После чего этот класс можно спокойно использовать для текстового поля у сообщений, заменив им базовый CharField. Что гарантирует безопасность и надежность для пользователей.
Взять тот же поиск или фильтрацию: если данные зашифрованы на клиенте, реализовать их силами сервера или силами админа в БД становится технически невозможно. Именно поэтому техпроекты со сложной архитектурой выбирают второй способ — шифрование на стороне сервера (At Rest), чтобы сохранить контроль над собственной архитектурой. Безусловно, E2EE обеспечивает максимальную приватность, но платить за неё приходится огромным усложнением разработки и потерей гибкости.
Ну и, пожалуй, самое приятное — это логика работы Маркета. Здесь разработка шла по довольно понятному сценарию, который можно описать классическими методами записи и выборки данных из БД. Всё устроено прозрачно: на бэкенде отрабатывают триггеры и проверки, определяющие, будет ли опубликован конкретный товар и соответствует ли он правилам платформы.
Намного интереснее устроены алгоритмы ранжирования, или, проще говоря, лента листингов. Для её формирования нужен полноценный перебор и скоринг данных. Система начисляет определенные баллы за каждый кейс: например, учитывается дата публикации, (триггер для моментального взлета листинга вверх ), количество просмотров, количество лайков, количество заказов на объявление. На основе финального веса товара 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. Это бета‑версия, поэтому я готов к любой конструктивной критике в свой адрес.
