Distributed Realtime Service

This guide explains how to implement a distributed, globally-scalable realtime subscription service using Redis Cluster. This architecture allows you to have multiple realtime service instances across different regions while maintaining event consistency.

Architecture Overview

┌─────────────────┐     ┌─────────────────┐     ┌─────────────────┐
│   VSKI Master   │────▶│  Redis Cluster  │◀────│  Realtime Svc   │
│   (Webhooks)    │     │   (Event Bus)   │     │  (Instance 1)   │
└─────────────────┘     └─────────────────┘     └─────────────────┘
                               │                        │
                               │                ┌─────────────────┐
                               └───────────────▶│  Realtime Svc   │
                                                │  (Instance N)   │
                                                └─────────────────┘

Key Components:

VSKI Master - Emits webhooks on record changes (create/update/delete)
Redis Cluster - Acts as distributed event bus with pub/sub
Realtime Service - Multiple instances that:
- Accept WebSocket connections from clients
- Subscribe to Redis channels
- Distribute events to connected clients

Implementation

1. Project Setup

Create a new Python project:

mkdir realtime-service
cd realtime-service
python -m venv .venv
source .venv/bin/activate
pip install aiohttp redis pyjwt

Pin the dependencies in requirements.txt:

aiohttp>=3.9
redis>=5.0
PyJWT>=2.8

2. Create the Realtime Service

import asyncio
import json
import os
import time
import uuid
from dataclasses import dataclass, field
from typing import Any

from aiohttp import web, WSMsgType
from redis.asyncio import Redis

from auth import verify_token  # implement it; see Security below

# Configuration
JWT_SECRET = os.environ.get("JWT_SECRET", "your-jwt-secret")
REDIS_URL = os.environ.get("REDIS_URL", "redis://localhost:6379")
PORT = int(os.environ.get("PORT", "8080"))

redis = Redis.from_url(REDIS_URL, decode_responses=True)

# Event storage (for replay): collection -> ordered events
EVENT_TTL = 60 * 60 * 1000  # 1 hour, in ms
event_store: dict[str, list[dict[str, Any]]] = {}


@dataclass
class Client:
    id: str
    ws: web.WebSocketResponse
    # collection -> {"last_id": str | None}; last_id is the resume cursor
    subscriptions: dict[str, dict[str, Any]] = field(default_factory=dict)


# WebSocket -> client
clients: dict[web.WebSocketResponse, Client] = {}


def generate_event_id() -> str:
    return f"{int(time.time() * 1000)}-{uuid.uuid4().hex[:8]}"


# Store an event for replay, dropping entries older than EVENT_TTL
def store_event(collection: str, event: dict[str, Any]) -> None:
    now = time.time() * 1000
    events = [e for e in event_store.get(collection, []) if now - e["timestamp"] < EVENT_TTL]
    events.append(event)
    event_store[collection] = events


# Events after last_id; empty when last_id is unknown (the client fell
# too far behind and should resync)
def get_events_since(collection: str, last_id: str) -> list[dict[str, Any]]:
    events = event_store.get(collection, [])
    for i, event in enumerate(events):
        if event["id"] == last_id:
            return events[i + 1:]
    return []


# Replay buffered events to one subscriber
async def send_events_to_client(client: Client, collection: str) -> None:
    sub = client.subscriptions.get(collection)
    if not sub:
        return

    if sub["last_id"]:
        events = get_events_since(collection, sub["last_id"])
    else:
        events = event_store.get(collection, [])
    if not events:
        return

    await client.ws.send_str(json.dumps({
        "type": "EVENT",
        "collection": collection,
        "events": [{"id": e["id"], "data": e["data"]} for e in events],
    }))
    sub["last_id"] = events[-1]["id"]


# Handle webhooks from the VSKI master. Non-POST requests get a 405 from
# the router (the route is registered via web.post).
# Payload shape: database, collection, id, action, timestamp, record.
async def handle_webhook(request: web.Request) -> web.Response:
    # Optional: verify the webhook JWT
    auth_header = request.headers.get("Authorization")
    if auth_header and auth_header.startswith("Bearer "):
        if verify_token(auth_header[7:]) is None:
            return web.Response(status=401, text="Unauthorized")

    data = await request.json()
    event = {
        "id": generate_event_id(),
        "data": {"action": data["action"], "record": data["record"]},
        "timestamp": int(time.time() * 1000),
    }

    # Every instance (this one included) stores and fans out via the
    # Redis bridge below, so there is a single delivery path
    await redis.publish(f"events:{data['collection']}", json.dumps(event))
    return web.json_response({"success": True})


async def handle_client_message(client: Client, msg: dict[str, Any]) -> None:
    if msg.get("type") == "SUBSCRIBE":
        collection = msg.get("collection", "")
        last_id = msg.get("lastId") or msg.get("lastIdArg")

        client.subscriptions[collection] = {
            "last_id": None if last_id == "$" else last_id,
        }

        await send_events_to_client(client, collection)
        await client.ws.send_str(json.dumps({
            "type": "SUBSCRIBED",
            "collection": collection,
            "last_id": client.subscriptions[collection]["last_id"] or "0",
        }))

    elif msg.get("type") == "UNSUBSCRIBE":
        client.subscriptions.pop(msg.get("collection", ""), None)

    elif msg.get("type") == "ACK":
        sub = client.subscriptions.get(msg.get("collection", ""))
        if sub is not None:
            sub["last_id"] = msg.get("id")


# Handle a WebSocket subscriber
async def handle_websocket(request: web.Request) -> web.StreamResponse:
    token = request.query.get("auth")
    if not token:
        return web.Response(status=401, text="Unauthorized")

    ws = web.WebSocketResponse()
    await ws.prepare(request)

    client = Client(id=str(uuid.uuid4()), ws=ws)
    clients[ws] = client
    print(f"Client {client.id} connected")

    async for msg in ws:
        if msg.type == WSMsgType.TEXT:
            await handle_client_message(client, json.loads(msg.data))
        elif msg.type == WSMsgType.ERROR:
            break

    clients.pop(ws, None)
    print(f"Client {client.id} disconnected")
    return ws


# Bridge Redis pub/sub to connected WebSocket clients
async def subscribe_to_redis() -> None:
    pubsub = redis.pubsub()  # dedicated connection from the pool
    await pubsub.psubscribe("events:*")
    try:
        async for message in pubsub.listen():
            if message["type"] != "pmessage":
                continue  # skip subscription confirmations

            channel = message["channel"]
            if not channel.startswith("events:"):
                continue

            collection = channel[len("events:"):]
            event = json.loads(message["data"])

            store_event(collection, event)

            # Broadcast to subscribed clients
            payload = json.dumps({
                "type": "EVENT",
                "collection": collection,
                "events": [{"id": event["id"], "data": event["data"]}],
            })
            for ws, client in list(clients.items()):
                if collection not in client.subscriptions:
                    continue
                try:
                    await ws.send_str(payload)
                except ConnectionResetError:
                    clients.pop(ws, None)
    finally:
        await pubsub.aclose()


async def handle_health(request: web.Request) -> web.Response:
    return web.json_response({"status": "ok", "clients": len(clients)})


async def on_startup(app: web.Application) -> None:
    app["redis_bridge"] = asyncio.create_task(subscribe_to_redis())


async def on_cleanup(app: web.Application) -> None:
    app["redis_bridge"].cancel()
    try:
        await app["redis_bridge"]
    except asyncio.CancelledError:
        pass
    await redis.aclose()


def create_app() -> web.Application:
    app = web.Application()
    app.add_routes([
        web.post("/webhook", handle_webhook),
        web.get("/api/realtime", handle_websocket),
        web.get("/health", handle_health),
    ])
    app.on_startup.append(on_startup)
    app.on_cleanup.append(on_cleanup)
    return app


if __name__ == "__main__":
    print(f"Realtime service running on port {PORT}")
    web.run_app(create_app(), port=PORT)

3. Configure VSKI Master Webhook

Create a collection with a trigger that sends events to your realtime service:

version: 1
database: default

collections:
  posts:
    type: base
    fields:
      - {name: title, type: text, required: true}
      - {name: content, type: text}
    options:
      triggers:
        - action: create
          url: https://realtime.example.com/webhook
          method: POST
          headers:
            Authorization: "Bearer ${REALTIME_JWT}"
        - action: update
          url: https://realtime.example.com/webhook
          method: POST
          headers:
            Authorization: "Bearer ${REALTIME_JWT}"
        - action: delete
          url: https://realtime.example.com/webhook
          method: POST
          headers:
            Authorization: "Bearer ${REALTIME_JWT}"

vski plan      # + collection default/posts (create base table with 2 fields)
vski apply     # creates it via a migration

4. Client Usage

import asyncio
from vski import VskiClient

user_token = "your-jwt-token"  # issued by the VSKI master


async def main():
    # Connect to the realtime service for subscriptions
    async with VskiClient("https://realtime.example.com") as realtime:
        realtime.set_token(user_token)

        def on_event(event):
            print(event["action"], event["record"])

        unsubscribe = realtime.collection("posts").subscribe(on_event)

        # Use master/replica for data operations
        async with VskiClient("https://master.example.com") as api:
            api.set_token(user_token)
            await api.collection("posts").create(
                {"title": "Hello", "content": "World"}
            )

        unsubscribe()

asyncio.run(main())

Docker Deployment

Dockerfile

FROM python:3.12-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY service.py .

ENV PORT=8080
ENV JWT_SECRET=your-jwt-secret
ENV REDIS_URL=redis://redis:6379

EXPOSE 8080
CMD ["python", "service.py"]

docker-compose.yml

version: "3.8"

services:
  redis:
    image: redis:7-alpine
    volumes:
      - redis_data:/data
    command: redis-server --appendonly yes

  realtime:
    build: .
    ports:
      - "8080:8080"
    environment:
      - JWT_SECRET=${JWT_SECRET}
      - REDIS_URL=redis://redis:6379
    depends_on:
      - redis
    deploy:
      replicas: 3

volumes:
  redis_data:

Redis Cluster (Production)

from redis.asyncio.cluster import RedisCluster

# One seed node is enough: the client discovers the full topology.
# Redis broadcasts PUBLISH cluster-wide, so an instance subscribed on
# any node sees every event.
redis = RedisCluster(
    host="redis-1.example.com",
    port=6379,
    decode_responses=True,
)

Load Balancer (nginx)

upstream realtime {
    least_conn;
    server realtime-1:8080;
    server realtime-2:8080;
    server realtime-3:8080;
}

server {
    listen 80;
    server_name realtime.example.com;

    location /api/realtime {
        proxy_pass http://realtime;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";
        proxy_set_header Host $host;
    }

    location /webhook {
        proxy_pass http://realtime;
    }
}

Security

JWT Verification (Production)

For production, use a proper JWT library:

# auth.py - using PyJWT
import os

import jwt

JWT_SECRET = os.environ.get("JWT_SECRET", "your-jwt-secret")


def verify_token(token: str) -> dict | None:
    try:
        # Pin the algorithm explicitly; never honor the token header
        return jwt.decode(token, JWT_SECRET, algorithms=["HS256"])
    except jwt.PyJWTError:
        return None

Rate Limiting

import time

rate_limiter: dict[str, list[float]] = {}


def check_rate_limit(ip: str, limit: int = 100, window_ms: int = 60_000) -> bool:
    now = time.time() * 1000
    requests = [t for t in rate_limiter.get(ip, []) if now - t < window_ms]
    if len(requests) >= limit:
        rate_limiter[ip] = requests
        return False
    requests.append(now)
    rate_limiter[ip] = requests
    return True


# In an aiohttp handler, key the window by the peer address:
#
#     if not check_rate_limit(request.remote):
#         return web.Response(status=429, text="Too many requests")

Monitoring

async def handle_health(request: web.Request) -> web.Response:
    return web.json_response({
        "status": "ok",
        "clients": len(clients),
        "collections": len(event_store),
        "redis": await redis.ping(),
    })

Multi-Region Setup

US Region:
  - Redis Cluster (US)
  - Realtime Service x3
  - Load Balancer

EU Region:
  - Redis Cluster (EU)
  - Realtime Service x3
  - Load Balancer

Cross-region sync via Redis replication or Kafka

Best Practices

Use same JWT secret as VSKI master for token compatibility
Authenticate webhook triggers - send the shared secret in the trigger’s Authorization header
Use Redis Cluster for production high availability
Deploy close to users for low latency
Monitor Redis memory for event storage