Designing a Notification System: From Naive Proposal to Production Architecture

Client                    Server
  │                          │
  │══ TCP Connection ════════│
  │  ┌─ Stream 1 (GET /api) ─┤
  │  ├─ Stream 2 (GET /img) ─┤  ← all from the SAME client
  │  └─ Stream 3 (SSE /notif)┤     inside ONE TCP connection
  │                          │

Event Source (any microservice)
        │
        ▼
Message Broker (RabbitMQ)
        │
        ▼
Notification Worker (Celery)
    ├── Persists to DB (audit log, offline inbox)
    └── Publishes to Redis Stream (key: notif:user:{id})
                │
                ▼
        SSE Server (FastAPI + async)
        └── Each client connection reads from its own Redis Stream
            └── Streams events over HTTP/2 SSE to browser

# ❌ DB-polling inside SSE — moves the problem, doesn't solve it
async def event_generator():
    while True:
        result = await db.execute(
            "SELECT * FROM notifications WHERE user_id=:id AND seen=false",
            {"id": current_user["user_id"]}
        )
        rows = result.fetchall()
        if rows:
            yield f"data: {rows}\n\n"
        await asyncio.sleep(1)

# ✅ Redis Streams — pure push, zero idle queries
async def event_generator():
    while True:
        messages = await redis_client.xread(
            {stream_key: cursor},
            block=5000  # sleeps here with zero CPU until data arrives
        )
        if messages:
            yield f"data: ...\n\n"

# Writer (Celery worker)
await redis_client.xadd(f"notif:user:{user_id}", payload)
 
# Reader (SSE server) — blocks until new message, replays on reconnect
messages = await redis_client.xread(
    {f"notif:user:{user_id}": last_seen_id},
    block=5000  # sleeps zero CPU until data arrives
)

# user_service/routers/timesheet.py
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from pydantic import BaseModel
from typing import List
import aio_pika, json
 
router = APIRouter()
 
class BulkRejectRequest(BaseModel):
    timesheet_ids: List[int]
    rejection_reason: str
 
@router.post("/timesheets/bulk-reject")
async def bulk_reject_timesheets(
    payload: BulkRejectRequest,
    db: AsyncSession = Depends(get_db),
    current_user: dict = Depends(require_role("Project Manager"))
):
    # 1. Single DB transaction for all 20
    async with db.begin():
        result = await db.execute(
            """
            UPDATE timesheets
            SET status = 'REJECTED',
                rejection_reason = :reason,
                rejected_by = :pm_id,
                rejected_at = NOW()
            WHERE id = ANY(:ids) AND status = 'PENDING'
            RETURNING id, employee_id
            """,
            {
                "reason": payload.rejection_reason,
                "pm_id": current_user["user_id"],
                "ids": payload.timesheet_ids
            }
        )
        updated = result.fetchall()
 
    if not updated:
        raise HTTPException(400, "No timesheets updated")
 
    # 2. ONE message to RabbitMQ — not 20
    affected = [
        {"timesheet_id": row.id, "employee_id": row.employee_id}
        for row in updated
    ]
 
    connection = await aio_pika.connect_robust("amqp://guest:guest@localhost/")
    async with connection:
        channel = await connection.channel()
        await channel.default_exchange.publish(
            aio_pika.Message(
                body=json.dumps({
                    "event": "TIMESHEET_BULK_REJECTED",
                    "pm_id": current_user["user_id"],
                    "pm_name": current_user["employee_name"],
                    "reason": payload.rejection_reason,
                    "affected_employees": affected
                }).encode()
            ),
            routing_key="notifications"
        )
 
    return {"rejected_count": len(updated)}

# notification_service/tasks.py
from celery import Celery
import redis.asyncio as redis
import json, asyncio
 
celery = Celery("notifications", broker="amqp://guest:guest@localhost/")
redis_client = redis.Redis(host="localhost", port=6379)
 
@celery.task(acks_late=True)  # acknowledge AFTER processing, not before
def handle_bulk_rejection(message_body: str):
    data = json.loads(message_body)
    asyncio.run(_fan_out(data))
 
async def _fan_out(data: dict):
    affected = data["affected_employees"]
 
    # Fan out to each employee's personal Redis stream
    pipe = redis_client.pipeline()
 
    for emp in affected:
        stream_key = f"notif:user:{emp['employee_id']}"
        pipe.xadd(
            stream_key,
            {
                "type": "TIMESHEET_REJECTED",
                "timesheet_id": emp["timesheet_id"],
                "reason": data["reason"],
                "rejected_by": data["pm_name"],
                "message": f"Your timesheet was rejected by {data['pm_name']}: {data['reason']}"
            },
            maxlen=100  # keep last 100 notifications per user
        )
 
    await pipe.execute()  # all 20 writes in ONE Redis round-trip

# notification_service/routers/sse.py
from fastapi import APIRouter
from fastapi.responses import StreamingResponse
import redis.asyncio as redis
import json, asyncio
 
router = APIRouter()
redis_client = redis.Redis(host="localhost", port=6379)
 
@router.get("/notifications/stream")
async def notification_stream(
    current_user: dict = Depends(get_current_user),
    last_id: str = "0"  # client sends last seen ID on reconnect
):
    async def event_generator():
        stream_key = f"notif:user:{current_user['user_id']}"
        cursor = last_id
 
        await redis_client.sadd("online_users", current_user["user_id"])
 
        try:
            while True:
                messages = await redis_client.xread(
                    {stream_key: cursor},
                    block=5000,  # sleep up to 5s, wake immediately if data arrives
                    count=10     # deliver max 10 at a time (burst control)
                )
 
                if not messages:
                    # Heartbeat — detects dead connections within 5 seconds
                    yield ": heartbeat\n\n"
                    continue
 
                for stream, events in messages:
                    for msg_id, fields in events:
                        cursor = msg_id  # advance cursor
                        yield f"id: {msg_id.decode()}\n"
                        yield f"data: {json.dumps({k.decode(): v.decode() for k, v in fields.items()})}\n\n"
 
        except asyncio.CancelledError:
            pass  # client disconnected cleanly
        finally:
            # Always runs — keeps online count accurate
            await redis_client.srem("online_users", current_user["user_id"])
 
    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no"  # critical — tells nginx not to buffer SSE
        }
    )

// hooks/useNotifications.js
import { useEffect, useRef, useCallback } from "react";
import { useNotificationStore } from "../store/notificationStore";
 
export function useNotifications() {
  const eventSourceRef = useRef(null);
  const lastIdRef = useRef(localStorage.getItem("notif_last_id") || "0");
  const addNotification = useNotificationStore(s => s.addNotification);
 
  const connect = useCallback(() => {
    const url = `/api/notifications/stream?last_id=${lastIdRef.current}`;
    const es = new EventSource(url);
 
    es.onmessage = (event) => {
      const data = JSON.parse(event.data);
 
      // Persist last seen ID — catches up on missed notifications after reconnect
      if (event.lastEventId) {
        lastIdRef.current = event.lastEventId;
        localStorage.setItem("notif_last_id", event.lastEventId);
      }
 
      addNotification(data);
 
      if (data.type === "TIMESHEET_REJECTED") {
        showToast(`Timesheet rejected: ${data.reason}`, "error");
      }
    };
 
    es.onerror = () => {
      es.close();
      // Auto-reconnect — passes last_id so no notifications missed
      setTimeout(connect, 3000);
    };
 
    eventSourceRef.current = es;
  }, [addNotification]);
 
  useEffect(() => {
    connect();
    return () => eventSourceRef.current?.close();
  }, [connect]);
}

// store/notificationStore.js (Zustand)
import { create } from "zustand";
 
export const useNotificationStore = create((set) => ({
  notifications: [],
  unreadCount: 0,
  addNotification: (notif) => set(state => ({
    notifications: [notif, ...state.notifications].slice(0, 20),
    unreadCount: state.unreadCount + 1
  })),
  markAllRead: () => set({ unreadCount: 0 })
}));

# Finite stream — sends 5 things and closes
async def event_generator():
    for i in range(5):
        yield f"data: message {i}\n\n"
        await asyncio.sleep(1)
# Connection closes after 5 messages

Python coroutine hits: await redis_client.xread(block=5000)
        │
        ▼
Python event loop parks the coroutine (no CPU consumed)
Monitors TCP socket via OS (epoll/kqueue)
Runs other coroutines freely
        │
        ▼
Redis server receives XREAD BLOCK 5000
Internally sets a 5000ms server-side timer
Does NOT respond immediately — waits for:
    ├── New message arrives → responds immediately (could be T=1s)
    └── 5000ms expires     → responds with empty reply
        │
        ▼
TCP socket receives Redis response
OS signals event loop: "data available"
Event loop wakes the coroutine
await returns → coroutine resumes

while True:
    messages = await redis_client.xread({...}, block=0)  # returns instantly with nothing
    # immediately loops back — 50K coroutines spinning = server melts

T=0s   User loses internet (no TCP FIN sent)
T=0s   Coroutine sleeping in xread...
T=5s   xread returns None (5s elapsed)
T=5s   Coroutine tries: yield ": heartbeat\n\n"
       OS detects broken pipe
       BrokenPipeError raised
T=5s   Coroutine dies → memory freed

T=0s    User opens browser → SSE connects → coroutine starts
T=0s    xread starts blocking (5s timer starts in Redis)
T=3s    PM rejects timesheet → Celery writes to Redis stream
T=3s    xread immediately wakes up (timer irrelevant, data arrived)
T=3s    notification yielded → browser receives it
T=3s    xread blocks again (new 5s timer starts)
T=8s    nothing arrived → xread returns None after 5s
T=8s    heartbeat yields ": heartbeat\n\n" → browser ignores it
T=8s    xread blocks again (new 5s timer starts)

T=0s   Coroutine blocking in xread
T=2s   User loses internet — no TCP FIN sent, FastAPI has NO idea
T=5s   xread returns None (5s elapsed)
T=5s   Coroutine tries to yield heartbeat to dead socket
       OS raises BrokenPipeError
T=5s   Coroutine dies → memory freed → online count decremented

# All event types go into the SAME stream — type differentiates them
await redis_client.xadd(
    f"notif:user:{user_id}",  # always same key
    {
        "type": "TIMESHEET_REJECTED",  # differentiator
        "payload": json.dumps(data)
    }
)

es.onmessage = (event) => {
    const data = JSON.parse(event.data);
    switch(data.type) {
        case "TIMESHEET_REJECTED": showTimesheetAlert(data); break;
        case "TASK_ASSIGNED":      updateTaskBoard(data);    break;
        case "LEAVE_APPROVED":     showLeaveStatus(data);    break;
    }
}

redis_client = redis.Redis(
    host="localhost",
    port=6379,
    max_connections=1000  # pool shared across all coroutines on this server
)

messages = await redis_client.xread(
    {stream_key: cursor},
    block=5000,
    count=10  # deliver max 10 at a time
)

async def event_generator(user_id, last_id):
    owner_key = f"notif:owner:{user_id}"
    my_id = f"server-{socket.gethostname()}-{uuid.uuid4()}"
 
    # Atomically claim ownership
    await redis_client.set(owner_key, my_id, ex=10)
    cursor = last_id
 
    try:
        while True:
            # Verify I still own this stream
            current_owner = await redis_client.get(owner_key)
            if current_owner.decode() != my_id:
                return  # I am the ghost — die cleanly
 
            messages = await redis_client.xread(
                {f"notif:user:{user_id}": cursor},
                block=5000,
                count=10
            )
 
            if not messages:
                # Refresh ownership TTL on heartbeat
                await redis_client.set(owner_key, my_id, ex=10)
                yield ": heartbeat\n\n"
                continue
 
            for stream, events in messages:
                for msg_id, fields in events:
                    cursor = msg_id
                    yield f"id: {msg_id.decode()}\n"
                    yield f"data: {json.dumps(...)}\n\n"
    finally:
        current_owner = await redis_client.get(owner_key)
        if current_owner and current_owner.decode() == my_id:
            await redis_client.delete(owner_key)

@router.post("/timesheets/bulk-reject")
async def bulk_reject(payload: BulkRejectRequest):
    # 1. DB write FIRST — source of truth
    async with db.begin():
        await db.execute("UPDATE timesheets SET status='REJECTED'...")
 
    # 2. ONLY after DB commit → publish to RabbitMQ
    await rabbitmq.publish("timesheet.rejected", payload)
 
    # 3. Return immediately — don't wait for notification delivery
    return {"rejected_count": 20}

async def _fan_out(data: dict):
    affected = data["affected_employees"]
 
    # 1. Redis writes — live delivery (milliseconds)
    pipe = redis_client.pipeline()
    for emp in affected:
        pipe.xadd(f"notif:user:{emp['employee_id']}", {...}, maxlen=100)
    await pipe.execute()
 
    # 2. DB writes — audit trail, history, read/unread (simultaneous, not triggered)
    await db.execute_many(
        "INSERT INTO notification_log (user_id, type, payload, created_at, read_at) VALUES (:user_id, :type, :payload, NOW(), NULL)",
        [{"user_id": emp["employee_id"], "type": "TIMESHEET_REJECTED", ...} for emp in affected]
    )

SELECT COUNT(*) FROM notification_log
WHERE user_id = 101 AND read_at IS NULL

# Notification created → increment
await redis_client.incr(f"notif:unread:{user_id}")
 
# User marks all read → reset
await redis_client.set(f"notif:unread:{user_id}", 0)
 
# Bell icon loads → O(1) read
count = await redis_client.get(f"notif:unread:{user_id}")

@router.patch("/notifications/{notif_id}/read")
async def mark_read(notif_id: str, current_user: dict = Depends(get_current_user)):
    await db.execute(
        "UPDATE notification_log SET read_at = NOW() WHERE id = :id AND user_id = :uid",
        {"id": notif_id, "uid": current_user["user_id"]}
    )
    await redis_client.decr(f"notif:unread:{current_user['user_id']}")
 
@router.patch("/notifications/read-all")
async def mark_all_read(current_user: dict = Depends(get_current_user)):
    await db.execute(
        "UPDATE notification_log SET read_at = NOW() WHERE user_id = :id AND read_at IS NULL",
        {"id": current_user["user_id"]}
    )
    await redis_client.set(f"notif:unread:{current_user['user_id']}", 0)

-- ❌ OFFSET pagination — scans and discards N rows every time
SELECT * FROM notification_log
WHERE user_id = 101
ORDER BY created_at DESC
LIMIT 20 OFFSET 10000  -- scans 10,020 rows, throws away 10,000

-- ✅ First page
SELECT * FROM notification_log
WHERE user_id = 101
ORDER BY created_at DESC
LIMIT 20;
 
-- ✅ Next page — pass last received created_at as cursor
SELECT * FROM notification_log
WHERE user_id = 101
AND created_at < '2024-01-15 10:30:00'  -- last seen timestamp
ORDER BY created_at DESC
LIMIT 20;

-- Composite index serves both the WHERE and ORDER BY in one scan
CREATE INDEX idx_notif_user_time
ON notification_log(user_id, created_at DESC);

useEffect(() => {
    // 1. Load recent from DB immediately — populates bell icon
    const recent = await fetch("/notifications/recent?limit=5")
    store.setInitial(recent)
 
    // 2. Simultaneously reconnect SSE — streams anything new
    connect()
}, [])