設計問題: SNSのタイムライン — Twitterのフィード設計

# ツイート投稿時 → 全フォロワーのタイムラインキャッシュに書き込む
class PushFanoutService
  def self.fanout(tweet)
    follower_ids = Follow.where(followee_id: tweet.user_id).pluck(:follower_id)
 
    follower_ids.each_slice(1000) do |batch|
      batch.each do |follower_id|
        $redis.lpush("timeline:#{follower_id}", tweet.id)
        $redis.ltrim("timeline:#{follower_id}", 0, 799)  # 最新800件のみ
      end
    end
  end
end
 
# メリット: 読み込みが高速（事前計算済み）
# デメリット: フォロワー100万人のユーザーが1ツイートで100万回書き込み

# タイムライン取得時 → フォロー中ユーザーのツイートをDBから取得
class PullTimelineService
  def self.fetch(user_id, limit: 20)
    followee_ids = Follow.where(follower_id: user_id).pluck(:followee_id)
 
    # フォロー200人 × SELECT = 非効率なN+1に近い処理
    Tweet.where(user_id: followee_ids)
         .order(created_at: :desc)
         .limit(limit)
  end
end
 
# メリット: 書き込み時の負荷がない
# デメリット: フォロー200人のツイートを毎回結合するので遅い

# app/services/fanout_service.rb
class FanoutService
  CELEBRITY_THRESHOLD = 10_000  # フォロワー数の閾値
 
  def self.fanout(tweet)
    user = User.find(tweet.user_id)
 
    if user.followers_count < CELEBRITY_THRESHOLD
      # 一般ユーザー: Pushモード
      PushFanoutJob.perform_later(tweet.id)
    else
      # 有名人: Pullモード用にマーキングのみ
      $redis.sadd("celebrity_tweets", tweet.id)
      $redis.expire("celebrity_tweets", 7.days.to_i)
    end
  end
end

// internal/fanout/worker.go
package fanout
 
import (
	"context"
	"sync"
)
 
type FanoutWorker struct {
	redis     RedisClient
	batchSize int
}
 
func (w *FanoutWorker) Fanout(ctx context.Context, tweetID int64, followerIDs []int64) error {
	batches := chunk(followerIDs, w.batchSize) // 1000件ずつバッチ
 
	var wg sync.WaitGroup
	errCh := make(chan error, len(batches))
 
	for _, batch := range batches {
		wg.Add(1)
		go func(ids []int64) {
			defer wg.Done()
			if err := w.pushToTimelines(ctx, tweetID, ids); err != nil {
				errCh <- err
			}
		}(batch)
	}
 
	wg.Wait()
	close(errCh)
 
	// エラー収集
	for err := range errCh {
		if err != nil {
			return err
		}
	}
	return nil
}
 
func (w *FanoutWorker) pushToTimelines(ctx context.Context, tweetID int64, followerIDs []int64) error {
	pipe := w.redis.Pipeline()
	for _, followerID := range followerIDs {
		key := fmt.Sprintf("timeline:%d", followerID)
		pipe.LPush(ctx, key, tweetID)
		pipe.LTrim(ctx, key, 0, 799)
	}
	_, err := pipe.Exec(ctx)
	return err
}
 
func chunk(ids []int64, size int) [][]int64 {
	var chunks [][]int64
	for size < len(ids) {
		ids, chunks = ids[size:], append(chunks, ids[0:size:size])
	}
	return append(chunks, ids)
}

# Sorted Setを使ったタイムライン管理
class TimelineWithZSet
  def self.add_tweet(follower_id, tweet_id, score)
    key = "tl:zset:#{follower_id}"
    $redis.zadd(key, score, tweet_id)
    $redis.zremrangebyrank(key, 0, -801)  # 古い800件を超えたら削除
  end
 
  # カーソルベースページネーション
  def self.fetch(follower_id, cursor: nil, limit: 20)
    key   = "tl:zset:#{follower_id}"
    max   = cursor ? "(#{cursor}" : '+inf'  # カーソル未満のスコアを取得
    ids   = $redis.zrevrangebyscore(key, max, '-inf', limit: [0, limit])
    next_cursor = ids.last ? $redis.zscore(key, ids.last) : nil
    { tweet_ids: ids, next_cursor: next_cursor }
  end
end

# オフセットベース（シンプルだが問題がある）
def fetch_offset(page:, per: 20)
  # 問題: page=2 取得中に新しいツイートが追加されると
  #       1ページ目の最後のツイートが2ページ目に流れて重複する
  tweet_ids = $redis.lrange("timeline:#{user_id}", page * per, (page + 1) * per - 1)
end
 
# カーソルベース（推奨）
def fetch_cursor(cursor: nil, limit: 20)
  # カーソル = 前回取得した最後のtweetのスコア（タイムスタンプ）
  # 新しいツイートが追加されても、カーソル位置が変わらない
  result = TimelineWithZSet.fetch(user_id, cursor: cursor, limit: limit)
  {
    tweets:      Tweet.where(id: result[:tweet_ids]).index_by(&:id),
    next_cursor: result[:next_cursor]
  }
end

# app/services/timeline_service.rb
class TimelineService
  CELEBRITY_THRESHOLD = 10_000
 
  def self.get_home_timeline(user_id, cursor: nil, limit: 20)
    # 1. Pushキャッシュからの取得（一般ユーザー分）
    push_tweets = fetch_push_timeline(user_id, cursor: cursor, limit: limit * 2)
 
    # 2. フォロー中のセレブリティを特定
    celebrity_ids = Follow.where(follower_id: user_id)
                          .joins(:followee)
                          .where('users.followers_count >= ?', CELEBRITY_THRESHOLD)
                          .pluck(:followee_id)
 
    # 3. セレブリティのツイートをPull
    celebrity_tweets = []
    if celebrity_ids.any?
      since = cursor ? Time.at(cursor.to_f / 1000) : 48.hours.ago
      celebrity_tweets = Tweet.where(user_id: celebrity_ids)
                               .where('created_at > ?', since)
                               .order(created_at: :desc)
                               .limit(limit)
    end
 
    # 4. マージして時系列ソート
    all_tweets = (push_tweets + celebrity_tweets)
                   .uniq(&:id)
                   .sort_by(&:created_at)
                   .reverse
                   .first(limit)
 
    all_tweets
  end
 
  private
 
  def self.fetch_push_timeline(user_id, cursor: nil, limit: 20)
    result = TimelineWithZSet.fetch(user_id, cursor: cursor, limit: limit)
    Tweet.where(id: result[:tweet_ids]).includes(:user)
  end
end

# lib/tasks/celebrity_classification.rake
namespace :celebrity do
  desc "フォロワー数に基づきユーザー分類を更新"
  task reclassify: :environment do
    # セレブリティ昇格（一般→セレブリティ）
    User.where(is_celebrity: false)
        .where('followers_count >= ?', FanoutService::CELEBRITY_THRESHOLD)
        .find_each do |user|
      user.update!(is_celebrity: true)
      # 既存のPushキャッシュを削除（移行）
      RemovePushCacheJob.perform_later(user.id)
    end
 
    # セレブリティ降格（セレブリティ→一般）
    User.where(is_celebrity: true)
        .where('followers_count < ?', FanoutService::CELEBRITY_THRESHOLD)
        .find_each do |user|
      user.update!(is_celebrity: false)
      # Pushキャッシュを再構築
      RebuildPushCacheJob.perform_later(user.id)
    end
  end
end

-- tweets テーブル（月単位パーティショニング）
CREATE TABLE tweets (
  id         BIGINT      NOT NULL,  -- Snowflake ID（時系列ソート可能）
  user_id    BIGINT      NOT NULL,
  content    VARCHAR(280) NOT NULL,
  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
) PARTITION BY RANGE (created_at);
 
-- インデックス設計
CREATE INDEX idx_tweets_user_id_created
  ON tweets (user_id, created_at DESC);  -- プロフィールタイムライン用
 
-- 月単位パーティション
CREATE TABLE tweets_2024_01 PARTITION OF tweets
  FOR VALUES FROM ('2024-01-01') TO ('2024-02-01');
 
-- follows テーブル（シャーディング候補）
CREATE TABLE follows (
  follower_id BIGINT NOT NULL,
  followee_id BIGINT NOT NULL,
  created_at  TIMESTAMPTZ NOT NULL DEFAULT NOW(),
  PRIMARY KEY (follower_id, followee_id)
);
CREATE INDEX idx_follows_followee ON follows(followee_id);
-- followee_idでの逆引き（フォロワー一覧取得）用

# app/jobs/tweet_index_job.rb
class TweetIndexJob < ApplicationJob
  queue_as :search_indexing
 
  def perform(tweet_id)
    tweet = Tweet.find(tweet_id)
 
    OPENSEARCH_CLIENT.index(
      index: "tweets-#{tweet.created_at.strftime('%Y-%m')}",
      id:    tweet.id,
      body:  {
        user_id:    tweet.user_id,
        content:    tweet.content,
        created_at: tweet.created_at.iso8601,
        # 検索用のメタデータ
        hashtags: extract_hashtags(tweet.content),
        mentions: extract_mentions(tweet.content)
      }
    )
  end
 
  private
 
  def extract_hashtags(content)
    content.scan(/#\w+/).map(&:downcase)
  end
 
  def extract_mentions(content)
    content.scan(/@\w+/).map { |m| m[1..] }  # @を除去
  end
end
 
# 検索クエリ例
def search_tweets(query, user_id:, cursor: nil)
  OPENSEARCH_CLIENT.search(
    index: 'tweets-*',
    body: {
      query: {
        bool: {
          must: [{ match: { content: query } }],
          filter: [
            # フォロー中のユーザーのみ
            { terms: { user_id: following_ids(user_id) } }
          ]
        }
      },
      sort: [{ created_at: { order: 'desc' } }],
      search_after: cursor ? [cursor] : nil,
      size: 20
    }
  )
end

Write Path（投稿フロー）:
  Client
  → ALB
  → Tweet API（ECS Fargate × 3）
  → Aurora PostgreSQL（永続化）
  → Amazon MSK（Kafka、ファンアウトキュー）
  → Fanout Worker（ECS Fargate × 10台）
  → ElastiCache Redis Cluster（タイムラインキャッシュ）
  → Amazon OpenSearch（全文検索インデックス）
 
Read Path（タイムライン取得フロー）:
  Client
  → ALB
  → Timeline API（ECS Fargate × 10）
  → ElastiCache Redis（キャッシュヒット → そのまま返す）
  → Aurora Reader Replica（キャッシュミス時のみ）
  + セレブリティツイートのリアルタイムマージ
 
Amazon MSK（Kafka）採用理由:
  - ファンアウトWorkerがスケールアウトしても確実にメッセージを配信
  - Consumerグループで並列処理
  - at-least-onceのメッセージ配信保証
  - 7日間のメッセージ保持（リプレイ可能）
 
Aurora PostgreSQL:
  Writer 1台 + Reader 3台（Multi-AZ）
  db.r6g.2xlarge（8vCPU, 64GB）
  接続数: PgBouncer経由でコネクションプール管理
 
ElastiCache Redis Cluster:
  cache.r6g.2xlarge × 6ノード（3シャード × 2レプリカ）
  合計メモリ: 312GB
  各ユーザーのタイムライン: 800件 × 8bytes ≒ 6.4KB
  → 100Mユーザー: 640GB（キャッシュに入りきらないのでLRU）

# app/middleware/timeline_metrics.rb
class TimelineMetrics
  FANOUT_LATENCY = Prometheus::Client::Histogram.new(
    :fanout_latency_seconds,
    docstring: 'Time to fanout tweet to all followers',
    buckets: [0.1, 0.5, 1.0, 5.0, 10.0, 30.0]
  )
  TIMELINE_CACHE_HIT = Prometheus::Client::Counter.new(
    :timeline_cache_hit_total, docstring: 'Timeline cache hits'
  )
  FANOUT_QUEUE_DEPTH = Prometheus::Client::Gauge.new(
    :fanout_queue_depth, docstring: 'MSK Fanout queue depth'
  )
end

# CloudWatch アラーム設定
アラーム一覧:
  FanoutLatencyHigh:
    メトリクス: p99(fanout_latency_seconds)
    閾値: 30秒超（ファンアウト遅延）
    アクション: Fanout Workerのスケールアウト
 
  MSKConsumerLag:
    メトリクス: kafka.consumer_lag
    閾値: 100,000メッセージ超
    アクション: PagerDuty通知 + Workerスケールアウト
 
  TimelineCacheHitRateLow:
    メトリクス: timeline_cache_hit / total_requests
    閾値: 80%未満
    アクション: Slack通知（キャッシュ設計の見直し検討）
 
  TimelineP99Latency:
    メトリクス: p99(timeline_fetch_latency)
    閾値: 100ms超
    アクション: PagerDuty通知