Design a Photo Sharing App (Instagram)

import io, json, time
import boto3, redis
from PIL import Image

r = redis.Redis()
s3 = boto3.client("s3", endpoint_url="https://<accountid>.r2.cloudflarestorage.com")
SIZES = {"thumb": 150, "feed": 1080}

def process(photo_id: str, key: str):
    raw = s3.get_object(Bucket="photos", Key=key)["Body"].read()
    img = Image.open(io.BytesIO(raw)).convert("RGB")  # decode once
    for name, width in SIZES.items():
        h = int(img.height * width / img.width)
        out = io.BytesIO()
        img.resize((width, h)).save(out, "WEBP", quality=82)
        variant_key = "photos/" + photo_id + "/" + name + ".webp"  # deterministic: retries overwrite
        s3.put_object(Bucket="photos", Key=variant_key, Body=out.getvalue(),
                      ContentType="image/webp", CacheControl="public, max-age=31536000, immutable")

while True:
    item = r.brpop("resize_queue", timeout=5)
    if not item:
        continue
    job = json.loads(item[1])
    try:
        process(job["photo_id"], job["s3_key"])
        r.lpush("fanout_queue", job["photo_id"])  # ready for feed fan-out
    except Exception:
        attempts = job.get("attempts", 0) + 1
        job["attempts"] = attempts
        target = "resize_dlq" if attempts >= 3 else "resize_queue"
        r.lpush(target, json.dumps(job))
        time.sleep(1)

async function getFeed(userId: string, cursor: bigint | null, limit = 20) {
  // 1. Precomputed ids from fan-out-on-write
  const pushedIds = (await redis.lrange("feed:" + userId, 0, 499))
    .map(BigInt)
    .filter((id) => cursor === null || id < cursor);

  // 2. Pull recent posts from followed celebrities (not fanned out)
  const celebs = await db.query(
    "SELECT f.followee_id FROM follows f JOIN users u ON u.id = f.followee_id " +
    "WHERE f.follower_id = $1 AND u.follower_count > 10000", [userId]);
  const celebIds: bigint[] = celebs.rows.length === 0 ? [] :
    (await db.query(
      "SELECT photo_id FROM photos WHERE user_id = ANY($1) AND status = 'ready' " +
      "AND ($2::bigint IS NULL OR photo_id < $2) ORDER BY photo_id DESC LIMIT $3",
      [celebs.rows.map((r: any) => r.followee_id), cursor, limit]
    )).rows.map((r: any) => BigInt(r.photo_id));

  // 3. Merge by id desc (snowflake ids sort by time) and hydrate
  const merged = [...new Set([...pushedIds, ...celebIds])]
    .sort((a, b) => (a > b ? -1 : 1))
    .slice(0, limit);
  const photos = await hydratePhotos(merged); // multi-get metadata + Redis counts
  return { photos, nextCursor: merged.length ? merged[merged.length - 1].toString() : null };
}

-- Truth: one row per (photo, user); re-likes are no-ops
INSERT INTO likes (photo_id, user_id, created_at)
VALUES ($1, $2, now())
ON CONFLICT (photo_id, user_id) DO NOTHING;

-- Batch applier: every 100 ms, collapse queued increments into one UPDATE per photo
UPDATE photos p
SET like_count = p.like_count + b.delta
FROM (VALUES ($1::bigint, $2::int)) AS b(photo_id, delta)
WHERE p.photo_id = b.photo_id;

-- Nightly reconciliation: repair drift from lost increments
UPDATE photos p
SET like_count = t.actual
FROM (SELECT photo_id, count(*) AS actual FROM likes GROUP BY photo_id) t
WHERE p.photo_id = t.photo_id AND p.like_count <> t.actual;