Проектирование Instagram

Проектирование системы, похожей на Instagram

Это комплексная система высокой нагрузки. Рассмотрим все слои архитектуры: от микросервисов до хранилища данных и кэширования.

1. Архитектура микросервисов

┌─────────────────────────────────────┐
│        API Gateway                  │
│      (Load Balancer, Auth)          │
└────────────┬────────────────────────┘
             │
    ┌────────┼────────┬──────────┬─────────┐
    │        │        │          │         │
┌───▼──┐ ┌──▼──┐ ┌───▼──┐ ┌────▼─┐ ┌──▼───┐
│User  │ │Feed │ │Post  │ │ Like │ │Follow│
│Service│ │Service│Service│Service│Service│
└──┬───┘ └──┬───┘ └───┬──┘ └────┬─┘ └──┬───┘
   │        │        │         │      │
   └────────┼────────┼─────────┼──────┘
            │        │         │
    ┌───────┴────────┴─────────┴──────┐
    │     PostgreSQL (масштабируемо)  │
    │  (Пользователи, посты, ребра)   │
    └───────────────────────────────┬──┘
                                    │
         ┌──────────┬───────────────┼──────────┬─────────┐
         │          │               │          │         │
    ┌────▼──┐  ┌────▼──┐  ┌────────▼──┐  ┌───▼──┐ ┌──▼──┐
    │Redis  │  │ S3/   │  │ Elasticsearch│ │Kafka │ │ CDN │
    │Cache  │  │Google │  │  (Search)    │ │Event │ │     │
    │       │  │Cloud  │  │              │ │Stream│ │     │
    └───────┘  │Storage│  └──────────────┘ └──────┘ └─────┘
               └───────┘

2. Сервисы и их ответственность

User Service:

from fastapi import FastAPI, Depends
from sqlalchemy import Column, String, DateTime, Integer
from datetime import datetime, timezone

app = FastAPI()

class User(Base):
    __tablename__ = "users"
    
    id = Column(String(36), primary_key=True)
    username = Column(String(100), unique=True, index=True)
    email = Column(String(255), unique=True, index=True)
    bio = Column(String(500))
    avatar_url = Column(String(2048))
    followers_count = Column(Integer, default=0, index=True)
    following_count = Column(Integer, default=0)
    created_at = Column(DateTime(timezone=True), default=datetime.now(timezone.utc))

@app.post("/api/v1/users/register")
async def register_user(username: str, email: str, password: str):
    """Регистрация пользователя."""
    # Хеширование пароля
    # Создание записи в БД
    # Отправка письма подтверждения
    pass

@app.get("/api/v1/users/{user_id}")
async def get_user_profile(user_id: str):
    """Получение профиля пользователя."""
    # Кэширование в Redis (TTL 1 час)
    pass

Post Service:

class Post(Base):
    __tablename__ = "posts"
    
    id = Column(String(36), primary_key=True)
    user_id = Column(String(36), ForeignKey('users.id'), index=True)
    caption = Column(String(2200))
    image_url = Column(String(2048))  # Ссылка на S3
    like_count = Column(Integer, default=0)
    comment_count = Column(Integer, default=0)
    created_at = Column(DateTime(timezone=True), default=datetime.now(timezone.utc), index=True)
    
    # Индекс для быстрого получения постов пользователя
    __table_args__ = (
        Index('idx_user_created', 'user_id', 'created_at'),
    )

@app.post("/api/v1/posts")
async def create_post(user_id: str, file: UploadFile, caption: str):
    """Загрузка фотографии."""
    # 1. Загружаем изображение в S3/Cloud Storage
    image_url = await upload_to_s3(file)
    
    # 2. Сохраняем метаданные в PostgreSQL
    post = Post(user_id=user_id, image_url=image_url, caption=caption)
    db.add(post)
    await db.commit()
    
    # 3. Отправляем событие в Kafka для обновления лент
    await kafka_producer.send("post_created", {"post_id": post.id, "user_id": user_id})
    
    return {"id": post.id, "image_url": image_url}

Follow Service:

class Follow(Base):
    __tablename__ = "follows"
    
    follower_id = Column(String(36), ForeignKey('users.id'), index=True, primary_key=True)
    following_id = Column(String(36), ForeignKey('users.id'), index=True, primary_key=True)
    created_at = Column(DateTime(timezone=True), default=datetime.now(timezone.utc))
    
    # Предотвращение дублей
    __table_args__ = (
        UniqueConstraint('follower_id', 'following_id'),
    )

@app.post("/api/v1/users/{user_id}/follow")
async def follow_user(follower_id: str, user_id: str):
    """Подписка на пользователя."""
    # 1. Создаём запись Follow
    follow = Follow(follower_id=follower_id, following_id=user_id)
    await db.add(follow)
    
    # 2. Инкрементируем счётчики
    await db.execute(
        update(User).where(User.id == user_id).values(
            followers_count=User.followers_count + 1
        )
    )
    await db.execute(
        update(User).where(User.id == follower_id).values(
            following_count=User.following_count + 1
        )
    )
    await db.commit()
    
    # 3. Инвалидируем кэш ленты
    await redis.delete(f"feed:{follower_id}")

Feed Service (самый сложный компонент):

class FeedService:
    
    async def get_feed(self, user_id: str, offset: int = 0, limit: int = 20):
        """Получение ленты новостей пользователя."""
        cache_key = f"feed:{user_id}:{offset}"
        
        # 1. Проверяем Redis кэш (горячие данные)
        cached = await redis.get(cache_key)
        if cached:
            return json.loads(cached)
        
        # 2. Получаем список people, на которых подписан пользователь
        following = await db.execute(
            select(Follow.following_id)
            .where(Follow.follower_id == user_id)
        )
        following_ids = [row[0] for row in following]
        
        # 3. Получаем посты от этих пользователей
        posts = await db.execute(
            select(Post)
            .where(Post.user_id.in_(following_ids))
            .order_by(Post.created_at.desc())
            .offset(offset)
            .limit(limit)
        )
        
        # 4. Обогащаем данные (likes, comments)
        feed_data = await self._enrich_posts(posts)
        
        # 5. Кэшируем на 5 минут
        await redis.setex(cache_key, 300, json.dumps(feed_data))
        
        return feed_data
    
    async def _enrich_posts(self, posts: List[Post]) -> List[dict]:
        """Добавляем дополнительные данные к постам."""
        enriched = []
        for post in posts:
            user = await self._get_user_cached(post.user_id)
            enriched.append({
                "id": post.id,
                "user": user,
                "caption": post.caption,
                "image_url": post.image_url,
                "like_count": post.like_count,
                "comment_count": post.comment_count,
                "created_at": post.created_at.isoformat()
            })
        return enriched
    
    async def _get_user_cached(self, user_id: str) -> dict:
        """Получение данных пользователя с кэшированием."""
        cache_key = f"user:{user_id}"
        cached = await redis.get(cache_key)
        
        if cached:
            return json.loads(cached)
        
        user = await db.get(User, user_id)
        user_data = {
            "id": user.id,
            "username": user.username,
            "avatar_url": user.avatar_url
        }
        
        await redis.setex(cache_key, 3600, json.dumps(user_data))
        return user_data

3. Выбор базы данных

PostgreSQL (основное хранилище):

• Транзакции (ACID) для операций со счётчиками
• Индексы для быстрого поиска
• Репликация для высокой доступности
• Partitioning для масштабирования

-- Partitioning по времени создания
CREATE TABLE posts (
    id UUID PRIMARY KEY,
    user_id UUID NOT NULL,
    caption VARCHAR(2200),
    created_at TIMESTAMPTZ NOT NULL
) PARTITION BY RANGE (created_at);

-- Партиции по месяцам
CREATE TABLE posts_2024_03 PARTITION OF posts
    FOR VALUES FROM ('2024-03-01') TO ('2024-04-01');

CREATE INDEX idx_user_created_2024_03 ON posts_2024_03(user_id, created_at DESC);

Redis (кэш и сессии):

• Кэширование профилей: TTL 1 час
• Кэширование лент: TTL 5 минут
• Session хранилище
• Rate limiting

Elasticsearch (поиск):

from elasticsearch import Elasticsearch

es_client = Elasticsearch(["localhost:9200"])

async def search_posts(query: str):
    """Полнотекстовый поиск по постам."""
    results = es_client.search(
        index="posts",
        body={
            "query": {
                "multi_match": {
                    "query": query,
                    "fields": ["caption", "user.username"]
                }
            }
        }
    )
    return results

4. Хранение изображений

AWS S3 / Google Cloud Storage:

import boto3
from botocore.exceptions import NoCredentialsError

class S3Service:
    def __init__(self):
        self.s3_client = boto3.client('s3')
        self.bucket_name = "instagram-clone-bucket"
    
    async def upload_image(self, file: UploadFile, user_id: str) -> str:
        """Загрузка изображения в S3."""
        try:
            # Генерируем уникальное имя файла
            file_key = f"posts/{user_id}/{uuid4()}_{file.filename}"
            
            # Загружаем с сжатием и оптимизацией
            await self.s3_client.put_object(
                Bucket=self.bucket_name,
                Key=file_key,
                Body=await file.read(),
                ContentType=file.content_type,
                CacheControl="max-age=86400",  # Кэш на 1 день
                Metadata={"user_id": user_id}
            )
            
            # Возвращаем URL с CDN
            return f"https://cdn.example.com/{file_key}"
        
        except NoCredentialsError:
            raise Exception("AWS credentials not found")
    
    async def generate_thumbnails(self, s3_url: str):
        """Генерируем разные размеры изображений."""
        # Lambda функция для обработки
        # Создаём: preview (200x200), thumbnail (400x400), large (1080x1080)
        pass

5. Генерирование ленты новостей

Подход 1: Push-модель (для небольших подписок)

# При создании поста отправляем событие всем подписчикам
async def on_post_created(post_id: str, user_id: str):
    # Получаем всех подписчиков
    followers = await db.execute(
        select(Follow.follower_id).where(Follow.following_id == user_id)
    )
    
    # Отправляем событие в Kafka для обновления лент
    for follower_id in followers:
        await kafka_producer.send(
            "feed_update",
            {"follower_id": follower_id, "post_id": post_id}
        )
        # Инвалидируем кэш ленты
        await redis.delete(f"feed:{follower_id}")

Подход 2: Pull-модель (рекомендуется)

# При запросе ленты получаем посты от подписок
async def get_feed_pull(user_id: str, offset: int = 0, limit: int = 20):
    # Получаем посты в порядке убывания времени создания
    posts = await db.execute(
        select(Post)
        .join(Follow, Post.user_id == Follow.following_id)
        .where(Follow.follower_id == user_id)
        .order_by(Post.created_at.desc())
        .offset(offset)
        .limit(limit)
    )
    return posts

Гибридный подход (лучший):

# Горячие посты (< 1 часа) в кэше через push
# Старые посты получаем through pull из БД
class HybridFeedService:
    async def get_feed(self, user_id: str, offset: int = 0, limit: int = 20):
        feed_items = []
        
        # 1. Получаем горячие посты из Redis (push-модель)
        hot_feed_key = f"hot_feed:{user_id}"
        hot_posts = await redis.lrange(hot_feed_key, 0, limit)
        feed_items.extend(hot_posts)
        
        # 2. Если не хватает, получаем из БД (pull-модель)
        if len(feed_items) < limit:
            remaining = limit - len(feed_items)
            cold_posts = await self._get_cold_posts(user_id, offset, remaining)
            feed_items.extend(cold_posts)
        
        return feed_items

6. Масштабирование системы

Горизонтальное масштабирование:

1. Multiple API Servers (stateless)
   ├─ Server 1
   ├─ Server 2
   └─ Server N

2. Load Balancer (nginx, HAProxy)
   └─ Round-robin / Least connections

3. Database Replication
   ├─ Primary (writes)
   ├─ Read Replica 1
   ├─ Read Replica 2
   └─ Read Replica N

4. Cache Layer (Redis Cluster)
   ├─ Shard 1
   ├─ Shard 2
   └─ Shard N

5. Message Queue (Kafka)
   ├─ Topic: post_created
   ├─ Topic: feed_update
   └─ Topic: notification

Оптимизация производительности:

from fastapi_limiter import FastAPILimiter
from prometheus_client import Counter, Histogram

# Rate limiting
limiter = FastAPILimiter(key_func=get_user_id)

@app.get("/api/v1/feed")
@limiter.limit("100/minute")
async def get_feed(user_id: str):
    pass

# Мониторинг
feed_requests = Counter('feed_requests_total', 'Total feed requests')
feed_latency = Histogram('feed_latency_seconds', 'Feed request latency')

@app.get("/api/v1/feed")
async def get_feed_monitored(user_id: str):
    with feed_latency.time():
        feed_requests.inc()
        return await get_feed(user_id)

7. Обработка особых случаев

Удаление постов:

@app.delete("/api/v1/posts/{post_id}")
async def delete_post(post_id: str, user_id: str):
    post = await db.get(Post, post_id)
    
    if post.user_id != user_id:
        raise PermissionDenied()
    
    # 1. Удаляем из БД (soft delete)
    await db.execute(
        update(Post).where(Post.id == post_id).values(deleted_at=datetime.now(UTC))
    )
    
    # 2. Удаляем изображение из S3
    await s3_service.delete_file(post.image_url)
    
    # 3. Инвалидируем кэши
    await redis.delete(f"post:{post_id}")
    await redis.delete(f"feed:*")  # Все ленты могут быть затронуты

Итоговая схема

Технологический стек:

• API: FastAPI с Uvicorn
• БД: PostgreSQL + Read Replicas + Partitioning
• Кэш: Redis Cluster
• Поиск: Elasticsearch
• Очереди: Kafka для асинхронных операций
• Хранилище: S3/Google Cloud Storage + CDN
• Контейнеризация: Docker + Kubernetes
• Мониторинг: Prometheus + Grafana

Эта архитектура выдерживает миллионы пользователей и миллиарды запросов в день.