短内容系统实现：从数据模型到 API 设计

// 热数据层 — posts collection (MongoDB)
const hotPostSchema = {
  _id: ObjectId,
  
  // 基础信息
  author: {
    user_id: ObjectId,
    username: String,
    display_name: String,
    avatar_version: Number  // 头像 CDN 版本，避免缓存穿透
  },
  
  // 内容核心（不可变）
  content: {
    text: String,           // 纯文本，不超过 500 字
    parsed_text: String,    // 解析后的 HTML（@用户、#话题、链接自动渲染）
    lang: String            // 语言检测，用于内容审核和推荐
  },
  
  // 多媒体（引用 CDN URL，非嵌入）
  media: [{
    type: String,           // image | video | audio | poll
    url: String,            // 原图/视频 URL
    variants: [{            // 响应式尺寸
      width: Number,
      height: Number,
      url: String
    }],
    meta: {                 // 媒体元信息
      duration: Number,     // 视频时长（秒）
      size_bytes: Number    // 文件大小
    }
  }],
  
  // 社交上下文
  interaction: {
    reply_to: ObjectId,     // 回复的目标帖子
    root_post: ObjectId,    // 话题根帖
    mentions: [{            // 提及的用户
      user_id: ObjectId,
      username: String,
      indices: [Number, Number]  // 在文本中的起止位置
    }],
    hashtags: [{            // 话题标签
      tag: String,
      indices: [Number, Number]
    }],
    links: [{               // 外链解析
      url: String,
      title: String,
      description: String,
      image: String,
      display_url: String
    }]
  },
  
  // 可见性控制
  visibility: {
    type: String,           // public | followers | mutuals | listed | direct
    reply_policy: String,   // everyone | followers | mentioned | nobody
    quote_policy: String    // allow | deny
  },
  
  // 计数（最终一致，由异步 Worker 维护）
  counters: {
    replies: Number,
    reposts: Number,
    quotes: Number,         // 引用转发（带评论的转发）
    likes: Number,
    bookmarks: Number,
    views: Number
  },
  
  // 时间戳
  created_at: Date,
  edited_at: Date,          // null 表示未编辑
  delete_scheduled_at: Date // 定时删除（阅后即焚模式）
};

// 用户资料变更时，更新 Redis 中的作者缓存
async function updateAuthorCache(userId, updates) {
  const newVersion = await incrAuthorVersion(userId);
  
  const cacheKey = `author:${userId}:v${newVersion}`;
  await redis.setex(cacheKey, 86400 * 30, JSON.stringify({
    user_id: userId,
    version: newVersion,
    ...updates
  }));
  
  // 旧版本保留 7 天后自动过期
  const oldVersion = newVersion - 1;
  if (oldVersion > 0) {
    await redis.expire(`author:${userId}:v${oldVersion}`, 86400 * 7);
  }
}

// 渲染 Feed 时，实时查询最新作者信息
async function resolveAuthors(posts) {
  const userIds = [...new Set(posts.map(p => p.author.user_id))];
  const versionMap = await redis.mget(
    userIds.map(id => `author:${id}:version`)
  );
  
  // 批量获取缓存
  const authorCache = await redis.mget(
    userIds.map((id, i) => `author:${id}:v${versionMap[i] || 1}`)
  );
  
  // 缓存缺失时回源数据库并回填
  // ...
  
  return posts.map(post => ({
    ...post,
    author: authorCache[post.author.user_id] || post.author
  }));
}

class FeedGenerator:
    def __init__(self):
        self.timeline_source = TimelineSource()
        self.trending_source = TrendingSource()
        self.interest_source = InterestSource()
        self.ads_source = AdsSource()
    
    async def generate(self, user_id: str, cursor: str = None) -> FeedResult:
        # 并行获取各来源
        results = await asyncio.gather(
            self.timeline_source.fetch(user_id, cursor),
            self.trending_source.fetch(user_id, limit=5),
            self.interest_source.fetch(user_id, limit=3),
            self.ads_source.fetch(user_id, limit=1)
        )
        
        timeline_posts, trending_posts, interest_posts, ads_posts = results
        
        # 按策略拼接
        feed = self._blend_posts(
            timeline_posts, trending_posts, interest_posts, ads_posts
        )
        
        # 去重（同一帖子可能出现在多个来源）
        feed = self._deduplicate(feed)
        
        # 个性化排序
        feed = self._personalize_rank(user_id, feed)
        
        # 生成下一页游标
        next_cursor = self._build_cursor(feed[-1]) if feed else None
        
        return FeedResult(posts=feed, cursor=next_cursor)
    
    def _blend_posts(self, timeline, trending, interest, ads) -> List[Post]:
        """按固定间隔拼接多个来源"""
        blended = []
        t_idx, tr_idx, i_idx, a_idx = 0, 0, 0, 0
        
        for position in range(50):  # 最多 50 条
            # 每 10 位插入 1 条广告
            if position > 0 and position % 10 == 0 and a_idx < len(ads):
                blended.append(ads[a_idx])
                a_idx += 1
                continue
            
            # 每 5 位插入 1 条趋势
            if position > 0 and position % 5 == 0 and tr_idx < len(trending):
                blended.append(trending[tr_idx])
                tr_idx += 1
                continue
            
            # 每 8 位插入 1 条兴趣探索
            if position > 0 and position % 8 == 0 and i_idx < len(interest):
                blended.append(interest[i_idx])
                i_idx += 1
                continue
            
            # 主要填充关注时间线
            if t_idx < len(timeline):
                blended.append(timeline[t_idx])
                t_idx += 1
            else:
                break
        
        return blended

import base64
import json

class CursorEncoder:
    @staticmethod
    def encode(post_id: str, score: float) -> str:
        """将游标编码为 URL-safe 字符串"""
        payload = json.dumps({"id": post_id, "s": score})
        return base64.urlsafe_b64encode(payload.encode()).decode().rstrip("=")
    
    @staticmethod
    def decode(cursor: str) -> dict:
        """解码游标"""
        padding = 4 - len(cursor) % 4
        if padding != 4:
            cursor += "=" * padding
        payload = base64.urlsafe_b64decode(cursor.encode())
        return json.loads(payload)

# 使用示例
# 下一页：/api/feed?cursor=eyJpZCI6InBvc3RfMTIzIiwicyI6MTcyNjk5ODQwMH0
# 游标包含最后一条帖子的 ID 和排序分数，确保分页边界精确

-- 个人互动状态（强一致）
CREATE TABLE user_interactions (
    user_id BIGINT NOT NULL,
    post_id BIGINT NOT NULL,
    liked BOOLEAN DEFAULT FALSE,
    reposted BOOLEAN DEFAULT FALSE,
    bookmarked BOOLEAN DEFAULT FALSE,
    created_at TIMESTAMPTZ DEFAULT NOW(),
    PRIMARY KEY (user_id, post_id)
);

-- 全局计数（最终一致，异步聚合）
CREATE TABLE post_counters (
    post_id BIGINT PRIMARY KEY,
    likes INT DEFAULT 0,
    replies INT DEFAULT 0,
    reposts INT DEFAULT 0,
    version INT DEFAULT 0  -- 乐观锁
);

class InteractionService:
    async def toggle_like(self, user_id: str, post_id: str) -> LikeResult:
        # 1. 写入个人互动状态（主库，强一致）
        async with self.db.transaction() as tx:
            result = await tx.fetchrow("""
                INSERT INTO user_interactions (user_id, post_id, liked)
                VALUES ($1, $2, TRUE)
                ON CONFLICT (user_id, post_id)
                DO UPDATE SET liked = NOT user_interactions.liked
                RETURNING liked
            """, user_id, post_id)
            
            is_liked = result['liked']
        
        # 2. 发送计数变更事件到 Kafka（异步）
        delta = 1 if is_liked else -1
        await self.kafka_producer.send('counter_events', {
            'post_id': post_id,
            'type': 'like',
            'delta': delta,
            'timestamp': time.time()
        })
        
        # 3. 乐观更新 Redis 缓存中的计数
        cache_key = f"post:{post_id}:counters"
        await redis.hincrby(cache_key, 'likes', delta)
        
        return LikeResult(liked=is_liked)
    
    async def get_user_interaction_state(self, user_id: str, post_ids: List[str]) -> dict:
        """批量查询用户对多条帖子的互动状态"""
        # 优先查 Redis 缓存
        keys = [f"ui:{user_id}:{pid}" for pid in post_ids]
        cached = await redis.mget(keys)
        
        missing = []
        for i, val in enumerate(cached):
            if val is None:
                missing.append(post_ids[i])
        
        # 缓存缺失时回源数据库
        if missing:
            rows = await self.db.fetch("""
                SELECT post_id, liked, reposted, bookmarked
                FROM user_interactions
                WHERE user_id = $1 AND post_id = ANY($2)
            """, user_id, missing)
            
            for row in rows:
                cache_key = f"ui:{user_id}:{row['post_id']}"
                state = json.dumps({
                    'liked': row['liked'],
                    'reposted': row['reposted'],
                    'bookmarked': row['bookmarked']
                })
                await redis.setex(cache_key, 86400, state)
        
        return {pid: json.loads(cached[i] or '{}') for i, pid in enumerate(post_ids)}

# counter_aggregator.py — Kafka 消费端
from kafka import KafkaConsumer
import asyncio

class CounterAggregator:
    def __init__(self):
        self.consumer = KafkaConsumer(
            'counter_events',
            group_id='counter-agg-v1',
            enable_auto_commit=False
        )
        self.buffer = defaultdict(lambda: defaultdict(int))
        self.last_flush = time.time()
    
    async def run(self):
        for message in self.consumer:
            event = json.loads(message.value)
            post_id = event['post_id']
            event_type = event['type']
            delta = event['delta']
            
            self.buffer[post_id][event_type] += delta
            
            # 定时或定量刷盘
            if (time.time() - self.last_flush > 5 or 
                sum(len(v) for v in self.buffer.values()) > 1000):
                await self._flush()
    
    async def _flush(self):
        for post_id, counters in self.buffer.items():
            # 使用乐观锁更新数据库
            while True:
                row = await self.db.fetchrow(
                    "SELECT likes, replies, reposts, version FROM post_counters WHERE post_id = $1",
                    post_id
                )
                
                if not row:
                    # 插入新记录
                    await self.db.execute("""
                        INSERT INTO post_counters (post_id, likes, replies, reposts, version)
                        VALUES ($1, $2, $3, $4, 1)
                        ON CONFLICT (post_id) DO NOTHING
                    """, post_id, counters.get('likes', 0), counters.get('replies', 0), counters.get('reposts', 0))
                    break
                
                new_likes = row['likes'] + counters.get('likes', 0)
                new_replies = row['replies'] + counters.get('replies', 0)
                new_reposts = row['reposts'] + counters.get('reposts', 0)
                new_version = row['version'] + 1
                
                result = await self.db.execute("""
                    UPDATE post_counters
                    SET likes = $1, replies = $2, reposts = $3, version = $4
                    WHERE post_id = $5 AND version = $6
                """, new_likes, new_replies, new_reposts, new_version, post_id, row['version'])
                
                if result == "UPDATE 1":
                    break  # 乐观锁成功
                # 失败则重试
        
        self.buffer.clear()
        self.last_flush = time.time()

{
  "mappings": {
    "properties": {
      "post_id": { "type": "keyword" },
      "content": {
        "type": "text",
        "analyzer": "standard",
        "fields": {
          "chinese": {
            "type": "text",
            "analyzer": "ik_max_word"
          }
        }
      },
      "hashtags": { "type": "keyword" },
      "mentions": { "type": "keyword" },
      "author_id": { "type": "keyword" },
      "author_username": { "type": "keyword" },
      "created_at": { "type": "date" },
      "likes": { "type": "integer" },
      "reposts": { "type": "integer" },
      "visibility": { "type": "keyword" },
      
      // 地理位置（可选）
      "location": { "type": "geo_point" }
    }
  },
  "settings": {
    "index": {
      "number_of_shards": 5,
      "number_of_replicas": 1,
      "refresh_interval": "5s"  // 近实时搜索
    }
  }
}

@app.get("/api/search")
async def search_posts(
    q: str,
    sort: str = "relevance",  # relevance | latest | popular
    filter: str = None,       # media | links | people
    since: datetime = None,
    until: datetime = None,
    cursor: str = None,
    limit: int = 20
):
    """全文搜索接口"""
    
    # 构建查询
    must_conditions = [
        {"multi_match": {
            "query": q,
            "fields": ["content^3", "content.chinese^3", "hashtags^2", "author_username"]
        }}
    ]
    
    if filter == "media":
        must_conditions.append({"exists": {"field": "media"}})
    elif filter == "links":
        must_conditions.append({"exists": {"field": "links"}})
    
    filter_conditions = [{"term": {"visibility": "public"}}]
    
    if since:
        filter_conditions.append({"range": {"created_at": {"gte": since}}})
    if until:
        filter_conditions.append({"range": {"created_at": {"lte": until}}})
    
    # 排序策略
    if sort == "latest":
        sort_clause = [{"created_at": "desc"}]
    elif sort == "popular":
        sort_clause = [
            {"likes": "desc"},
            {"reposts": "desc"}
        ]
    else:
        sort_clause = ["_score", {"created_at": "desc"}]
    
    # 游标分页
    search_after = None
    if cursor:
        search_after = json.loads(base64.b64decode(cursor))
    
    query = {
        "query": {
            "bool": {
                "must": must_conditions,
                "filter": filter_conditions
            }
        },
        "sort": sort_clause,
        "size": limit
    }
    
    if search_after:
        query["search_after"] = search_after
    
    response = await es.search(index="posts", body=query)
    
    hits = response["hits"]["hits"]
    posts = [hit["_source"] for hit in hits]
    
    # 生成下一页游标
    next_cursor = None
    if len(hits) == limit:
        last_sort = hits[-1]["sort"]
        next_cursor = base64.b64encode(json.dumps(last_sort).encode()).decode()
    
    return {
        "posts": posts,
        "total": response["hits"]["total"]["value"],
        "next_cursor": next_cursor
    }

// 成功响应
{
  "success": true,
  "data": { ... },           // 业务数据
  "meta": {
    "cursor": "...",         // 分页游标
    "has_more": true,
    "total_count": 1523
  }
}

// 错误响应
{
  "success": false,
  "error": {
    "code": "RATE_LIMITED",
    "message": "Too many requests, please try again later",
    "details": {
      "retry_after": 60       // 秒
    }
  }
}

# GET /api/v1/posts/123?fields=id,content,author(username,avatar),created_at
# GET /api/v1/feed?include=author,media,interaction.counters

class FieldSelector:
    def __init__(self, requested_fields: str):
        self.fields = self._parse(requested_fields) if requested_fields else None
    
    def filter(self, obj: dict) -> dict:
        if self.fields is None:
            return obj
        return {k: v for k, v in obj.items() if k in self.fields}
    
    def _parse(self, fields_str: str) -> set:
        """解析类似 GraphQL 的字段选择语法"""
        result = set()
        # 简单实现：逗号分隔顶层字段
        for field in fields_str.split(','):
            result.add(field.strip().split('.')[0])
        return result