Как работают Stories в Instagram: симулятор на Python

Зачем инженеру понимать устройство Stories?

Вы нажимаете кнопку «Добавить в историю» — и через секунду ваши подписчики видят фото. Со стороны это кажется магией. Но за простым интерфейсом скрывается сложная система: аутентификация, проверка прав, загрузка медиа, расчёт аудитории и управление временем жизни. В этой статье мы построим упрощённый симулятор backend Instagram Stories на Python, который визуализирует каждый этап. Вы поймёте, как работают очереди задач, распределённое хранение и фильтрация аудитории — и сможете применить эти принципы в своих проектах.

Архитектура симулятора: основные компоненты

Наш симулятор состоит из нескольких сервисов, каждый отвечает за свою часть workflow:

• Клиентское приложение (App) — отправляет запрос на загрузку сторис.
• API Gateway — проверяет аутентификацию пользователя.
• Сервис медиа (Media Service) — сохраняет изображение в хранилище.
• Сервис сторис (Story Service) — создаёт запись о сторис с метаданными.
• Сервис аудитории (Audience Service) — вычисляет, кто может видеть сторис.

Каждый сервис работает независимо, обмениваясь данными через очередь сообщений. Это позволяет обрабатывать миллионы запросов параллельно.

Код симулятора на Python

Ниже представлена упрощённая версия backend на Python. Мы используем асинхронные очереди для имитации микросервисной архитектуры.

import asyncio
import random
from dataclasses import dataclass
from typing import List, Optional

# --- Модели данных ---
@dataclass
class User:
    user_id: int
    username: str

@dataclass
class Story:
    story_id: int
    user_id: int
    media_url: str
    story_type: str  # 'normal' или 'close_friends'
    created_at: float
    expires_at: float

@dataclass
class Follower:
    user_id: int
    username: str
    is_blocked: bool = False
    is_hidden: bool = False
    is_close_friend: bool = False

# --- Сервисы ---
class AuthService:
    """Проверяет аутентификацию пользователя."""
    async def verify_token(self, token: str) -> Optional[User]:
        # Имитация проверки токена
        await asyncio.sleep(0.05)
        if token == "valid_token":
            return User(user_id=1, username="alice")
        return None

class MediaService:
    """Сохраняет медиафайл и возвращает URL."""
    async def upload_media(self, image_data: bytes) -> str:
        await asyncio.sleep(0.1)
        # Имитация сохранения в S3
        return f"https://media.cdn.com/stories/{random.randint(1000,9999)}.jpg"

class StoryService:
    """Создаёт запись о сторис в базе данных."""
    def __init__(self):
        self.stories = {}
        self.counter = 0

    async def create_story(self, user_id: int, media_url: str, story_type: str) -> Story:
        await asyncio.sleep(0.05)
        self.counter += 1
        story = Story(
            story_id=self.counter,
            user_id=user_id,
            media_url=media_url,
            story_type=story_type,
            created_at=asyncio.get_event_loop().time(),
            expires_at=asyncio.get_event_loop().time() + 86400  # 24 часа
        )
        self.stories[story.story_id] = story
        return story

class AudienceService:
    """Вычисляет список пользователей, которым доступна сторис."""
    async def compute_audience(self, story: Story, followers: List[Follower]) -> List[Follower]:
        await asyncio.sleep(0.1)
        if story.story_type == "close_friends":
            # Только близкие друзья
            return [f for f in followers if f.is_close_friend and not f.is_blocked]
        else:
            # Все подписчики, кроме заблокированных и скрытых
            return [f for f in followers if not f.is_blocked and not f.is_hidden]

# --- Оркестрация ---
class StoryUploadOrchestrator:
    def __init__(self):
        self.auth = AuthService()
        self.media = MediaService()
        self.story_service = StoryService()
        self.audience = AudienceService()

    async def upload_story(self, token: str, image_data: bytes, story_type: str, followers: List[Follower]):
        # Шаг 1: Аутентификация
        user = await self.auth.verify_token(token)
        if not user:
            raise PermissionError("Неверный токен")
        print(f"Пользователь {user.username} аутентифицирован")

        # Шаг 2: Загрузка медиа
        media_url = await self.media.upload_media(image_data)
        print(f"Медиа загружено: {media_url}")

        # Шаг 3: Создание сторис
        story = await self.story_service.create_story(user.user_id, media_url, story_type)
        print(f"Сторис #{story.story_id} создана")

        # Шаг 4: Расчёт аудитории
        audience = await self.audience.compute_audience(story, followers)
        print(f"Аудитория: {[f.username for f in audience]}")

        return story, audience

# --- Пример использования ---
async def main():
    # Подписчики
    followers = [
        Follower(2, "bob", is_close_friend=True),
        Follower(3, "charlie", is_blocked=True),
        Follower(4, "diana", is_hidden=True),
        Follower(5, "eve", is_close_friend=True),
    ]

    orchestrator = StoryUploadOrchestrator()
    story, audience = await orchestrator.upload_story(
        token="valid_token",
        image_data=b"fake_image_bytes",
        story_type="normal",
        followers=followers
    )
    print(f"Итог: сторис {story.story_id} видна {len(audience)} пользователям")

if __name__ == "__main__":
    asyncio.run(main())

Ключевые выводы

Мы разобрали, как работает загрузка сторис на уровне backend. Основные уроки:

1. Аутентификация — первый барьер: без проверки токена система уязвима.
2. Медиа-сервис отделён от логики: это позволяет масштабировать хранение независимо.
3. Расчёт аудитории — самая сложная часть: фильтрация по правам доступа требует быстрых алгоритмов.
4. Close Friends — элегантное расширение: меняется только логика аудитории, остальная система остаётся неизменной.

Теперь ваша очередь: возьмите этот код, добавьте реальную базу данных или очередь сообщений (например, Redis) и попробуйте запустить симуляцию с тысячами подписчиков. Вы увидите, как быстро растёт сложность.