Fan-out: write vs read

<?php

declare(strict_types=1);

namespace App\Timeline;

use Predis\ClientInterface as Redis;

/**
 * Threshold above which a user is treated as a "celebrity"
 * and we switch from push to pull for their posts.
 */
final class TimelineService
{
    private const CELEBRITY_THRESHOLD = 10_000;
    private const INBOX_LIMIT = 1000;
    private const FEED_PAGE_SIZE = 50;

    public function __construct(
        private readonly Redis $redis,
        private readonly PostRepository $posts,
        private readonly FollowRepository $follows,
    ) {}

    /**
     * On new post: push to follower inboxes, but skip if author is a celebrity.
     * Celebrity posts are pulled at read time from a shared cache.
     */
    public function onPostPublished(string $authorId, string $postId, int $createdAt): void
    {
        $followerCount = $this->follows->countFollowersOf($authorId);

        if ($followerCount > self::CELEBRITY_THRESHOLD) {
            // Push to a per-author "latest posts" cache, read by fans on feed load
            $this->cacheCelebrityPost($authorId, $postId, $createdAt);
            return;
        }

        // Regular user: fan-out on write to all followers
        $this->fanoutToFollowers($authorId, $postId, $createdAt);
    }

    private function cacheCelebrityPost(string $authorId, string $postId, int $ts): void
    {
        $key = "celeb:posts:$authorId";
        $this->redis->zadd($key, [$postId => $ts]);
        // Keep last 200 posts for celebrity
        $this->redis->zremrangebyrank($key, 0, -201);
    }

    /**
     * Fan-out to followers. For >10K followers this is batched.
     * In production run as a background job with chunking.
     */
    private function fanoutToFollowers(string $authorId, string $postId, int $ts): void
    {
        foreach ($this->follows->streamFollowers($authorId, chunkSize: 1000) as $chunk) {
            $pipe = $this->redis->pipeline();
            foreach ($chunk as $followerId) {
                $key = "inbox:$followerId";
                $pipe->zadd($key, [$postId => $ts]);
                $pipe->zremrangebyrank($key, 0, -self::INBOX_LIMIT - 1);
            }
            $pipe->execute();
        }
    }

    /**
     * Read feed: inbox (pushed posts) + celebrities pulled live.
     * Merge and return top N by timestamp.
     *
     * @return string[] post IDs
     */
    public function homeTimeline(string $userId): array
    {
        // 1. Pushed posts from regular users
        $inboxKey = "inbox:$userId";
        $pushed = $this->redis->zrevrange($inboxKey, 0, self::FEED_PAGE_SIZE - 1, 'WITHSCORES');

        // 2. Pull latest posts from celebrities user follows
        $celebs = $this->follows->getCelebrityFollows($userId);
        $celebPosts = [];
        foreach ($celebs as $celebId) {
            $key = "celeb:posts:$celebId";
            $items = $this->redis->zrevrange($key, 0, 20, 'WITHSCORES');
            foreach ($items as $postId => $score) {
                $celebPosts[$postId] = (int) $score;
            }
        }

        // 3. Merge: convert pushed to [id => score] and union
        $merged = [];
        foreach ($pushed as $postId => $score) {
            $merged[$postId] = (int) $score;
        }
        $merged += $celebPosts;

        // Sort by score desc, take top N
        arsort($merged);
        return array_slice(array_keys($merged), 0, self::FEED_PAGE_SIZE);
    }
}

package timeline

import (
	"context"
	"sort"

	"github.com/redis/go-redis/v9"
)

const (
	celebrityThreshold = 10_000
	inboxLimit         = 1000
	feedPageSize       = 50
)

type Service struct {
	rdb     *redis.Client
	follows FollowRepo
}

// OnPostPublished decides push vs celebrity cache based on follower count.
func (s *Service) OnPostPublished(ctx context.Context, authorID, postID string, ts int64) error {
	n, err := s.follows.CountFollowers(ctx, authorID)
	if err != nil {
		return err
	}
	if n > celebrityThreshold {
		return s.cacheCelebrity(ctx, authorID, postID, ts)
	}
	return s.fanoutToFollowers(ctx, authorID, postID, ts)
}

func (s *Service) cacheCelebrity(ctx context.Context, authorID, postID string, ts int64) error {
	key := "celeb:posts:" + authorID
	if err := s.rdb.ZAdd(ctx, key, redis.Z{Score: float64(ts), Member: postID}).Err(); err != nil {
		return err
	}
	// Keep last 200 posts
	return s.rdb.ZRemRangeByRank(ctx, key, 0, -201).Err()
}

// fanoutToFollowers streams follower IDs in chunks and pipelines ZADD per batch.
// For non-celebrity authors only. Runs as a background job in production.
func (s *Service) fanoutToFollowers(ctx context.Context, authorID, postID string, ts int64) error {
	stream, errCh := s.follows.StreamFollowers(ctx, authorID, 1000)
	for chunk := range stream {
		pipe := s.rdb.Pipeline()
		for _, followerID := range chunk {
			key := "inbox:" + followerID
			pipe.ZAdd(ctx, key, redis.Z{Score: float64(ts), Member: postID})
			pipe.ZRemRangeByRank(ctx, key, 0, -inboxLimit-1)
		}
		if _, err := pipe.Exec(ctx); err != nil {
			return err
		}
	}
	if err := <-errCh; err != nil {
		return err
	}
	return nil
}

type scoredPost struct {
	postID string
	score  int64
}

// HomeTimeline merges pushed inbox with pulled celebrity posts.
func (s *Service) HomeTimeline(ctx context.Context, userID string) ([]string, error) {
	inbox, err := s.rdb.ZRevRangeWithScores(ctx, "inbox:"+userID, 0, feedPageSize-1).Result()
	if err != nil {
		return nil, err
	}

	celebs, err := s.follows.CelebrityFollows(ctx, userID)
	if err != nil {
		return nil, err
	}

	merged := make(map[string]int64, len(inbox)+len(celebs)*20)
	for _, z := range inbox {
		merged[z.Member.(string)] = int64(z.Score)
	}
	// Pull recent posts from each celebrity (cached, shared across all fans)
	for _, celebID := range celebs {
		posts, err := s.rdb.ZRevRangeWithScores(ctx, "celeb:posts:"+celebID, 0, 20).Result()
		if err != nil {
			continue
		}
		for _, z := range posts {
			merged[z.Member.(string)] = int64(z.Score)
		}
	}

	list := make([]scoredPost, 0, len(merged))
	for id, score := range merged {
		list = append(list, scoredPost{id, score})
	}
	sort.Slice(list, func(i, j int) bool { return list[i].score > list[j].score })

	n := feedPageSize
	if len(list) < n {
		n = len(list)
	}
	out := make([]string, n)
	for i := 0; i < n; i++ {
		out[i] = list[i].postID
	}
	return out, nil
}

using Microsoft.Extensions.Logging;
using StackExchange.Redis;

namespace App.Timeline;

public sealed class Service(
    IConnectionMultiplexer redis,
    IFollowRepo follows,
    ILogger<Service> log)
{
    private const int CelebrityThreshold = 10_000;
    private const int InboxLimit = 1000;
    private const int FeedPageSize = 50;
    private const int CelebrityCacheSize = 200;

    private IDatabase Db => redis.GetDatabase();

    /// Decides push vs celebrity cache based on follower count.
    public async Task OnPostPublishedAsync(
        string authorId, string postId, long ts, CancellationToken ct)
    {
        var n = await follows.CountFollowersAsync(authorId, ct);
        if (n > CelebrityThreshold)
        {
            await CacheCelebrityAsync(authorId, postId, ts);
            return;
        }

        await FanoutToFollowersAsync(authorId, postId, ts, ct);
    }

    private async Task CacheCelebrityAsync(string authorId, string postId, long ts)
    {
        var key = (RedisKey)$"celeb:posts:{authorId}";
        await Db.SortedSetAddAsync(key, postId, ts);
        // Keep last 200 posts
        await Db.SortedSetRemoveRangeByRankAsync(key, 0, -CelebrityCacheSize - 1);
    }

    /// Streams follower IDs in chunks and pipelines ZADD per batch.
    /// For non-celebrity authors only. Runs as a background job in production.
    private async Task FanoutToFollowersAsync(
        string authorId, string postId, long ts, CancellationToken ct)
    {
        await foreach (var chunk in follows.StreamFollowersAsync(authorId, 1000, ct))
        {
            var batch = Db.CreateBatch();
            var pending = new List<Task>(chunk.Count * 2);

            foreach (var followerId in chunk)
            {
                var key = (RedisKey)$"inbox:{followerId}";
                pending.Add(batch.SortedSetAddAsync(key, postId, ts));
                pending.Add(batch.SortedSetRemoveRangeByRankAsync(key, 0, -InboxLimit - 1));
            }

            batch.Execute();
            await Task.WhenAll(pending);
        }
    }

    /// Merges the pushed inbox with pulled celebrity posts.
    public async Task<IReadOnlyList<string>> HomeTimelineAsync(
        string userId, CancellationToken ct)
    {
        var inbox = await Db.SortedSetRangeByRankWithScoresAsync(
            $"inbox:{userId}", 0, FeedPageSize - 1, Order.Descending);

        var celebs = await follows.CelebrityFollowsAsync(userId, ct);

        var merged = new Dictionary<string, long>(inbox.Length + celebs.Count * 20);
        foreach (var z in inbox)
        {
            merged[z.Element!] = (long)z.Score;
        }

        // Pull recent posts from each celebrity concurrently (cached, shared across fans)
        var pulls = celebs.Select(async celebId =>
        {
            try
            {
                return await Db.SortedSetRangeByRankWithScoresAsync(
                    $"celeb:posts:{celebId}", 0, 20, Order.Descending);
            }
            catch (RedisException e)
            {
                // A missing celebrity cache degrades the feed, it must not fail it
                log.LogWarning(e, "celebrity pull failed; celeb={CelebId}", celebId);
                return [];
            }
        });

        foreach (var posts in await Task.WhenAll(pulls))
        {
            foreach (var z in posts)
            {
                merged[z.Element!] = (long)z.Score;
            }
        }

        return merged
            .OrderByDescending(kv => kv.Value)
            .Take(FeedPageSize)
            .Select(kv => kv.Key)
            .ToList();
    }
}

import asyncio
import logging

import redis.asyncio as redis

log = logging.getLogger(__name__)

CELEBRITY_THRESHOLD = 10_000
INBOX_LIMIT = 1000
FEED_PAGE_SIZE = 50
CELEBRITY_CACHE_SIZE = 200

class Service:
    def __init__(self, client: redis.Redis, follows: FollowRepo) -> None:
        self._redis = client
        self._follows = follows

    async def on_post_published(self, author_id: str, post_id: str, ts: int) -> None:
        """Decide push vs celebrity cache based on follower count."""
        n = await self._follows.count_followers(author_id)
        if n > CELEBRITY_THRESHOLD:
            await self._cache_celebrity(author_id, post_id, ts)
            return
        await self._fanout_to_followers(author_id, post_id, ts)

    async def _cache_celebrity(self, author_id: str, post_id: str, ts: int) -> None:
        key = f"celeb:posts:{author_id}"
        async with self._redis.pipeline(transaction=True) as pipe:
            pipe.zadd(key, {post_id: ts})
            # Keep last 200 posts
            pipe.zremrangebyrank(key, 0, -CELEBRITY_CACHE_SIZE - 1)
            await pipe.execute()

    async def _fanout_to_followers(self, author_id: str, post_id: str, ts: int) -> None:
        """Stream follower IDs in chunks and pipeline ZADD per batch."""
        async for chunk in self._follows.stream_followers(author_id, 1000):
            async with self._redis.pipeline(transaction=False) as pipe:
                for follower_id in chunk:
                    key = f"inbox:{follower_id}"
                    pipe.zadd(key, {post_id: ts})
                    pipe.zremrangebyrank(key, 0, -INBOX_LIMIT - 1)
                await pipe.execute()

    async def home_timeline(self, user_id: str) -> list[str]:
        """Merge the pushed inbox with pulled celebrity posts."""
        inbox = await self._redis.zrevrange(
            f"inbox:{user_id}", 0, FEED_PAGE_SIZE - 1, withscores=True
        )
        celebs = await self._follows.celebrity_follows(user_id)

        merged: dict[str, int] = {member: int(score) for member, score in inbox}

        # Pull recent posts from each celebrity concurrently. asyncio gives real
        # parallelism here despite the GIL: these are network waits, not CPU work.
        results = await asyncio.gather(
            *(
                self._redis.zrevrange(f"celeb:posts:{cid}", 0, 20, withscores=True)
                for cid in celebs
            ),
            return_exceptions=True,
        )
        for celeb_id, posts in zip(celebs, results, strict=True):
            if isinstance(posts, BaseException):
                # A missing celebrity cache degrades the feed, it must not fail it
                log.warning("celebrity pull failed", extra={"celeb_id": celeb_id})
                continue
            merged.update({member: int(score) for member, score in posts})

        top = sorted(merged.items(), key=lambda kv: kv[1], reverse=True)
        return [post_id for post_id, _ in top[:FEED_PAGE_SIZE]]