csrd-realtime 0.5.28

WebSocket connection lifecycle, room broadcast, and routing for real-time services

csrd WebSocket Library

Status: Active implementation (v0.1.0) — core manager/router primitives available.

Real-time WebSocket connection lifecycle, room broadcast, and routing primitives for the csrd ecosystem.

Overview

This package provides reusable WebSocket infrastructure for building real-time services on FastAPI. It handles connection registration, room membership, message routing, and optional policy hooks — but does not define domain-specific protocols.

Design reference

• Design doc: ../../docs/WEBSOCKET_LIBRARY_DESIGN.md
• Implementation plan: ../../docs/WEBSOCKET_LIBRARY_IMPLEMENTATION_PLAN.md

Current API

from csrd.realtime import (
    ConnectionContext,
    ConnectionManager,
    MessageEnvelope,
    MessageRouter,
)

Status

Current milestone includes:

• ConnectionManager (connect/disconnect, room membership, room/global broadcast)
• MessageRouter (event dispatch + optional policy hooks)
• ConnectionContext and MessageEnvelope
• Typed errors for policy/validation/connection lookup

Minimal FastAPI wiring example

from fastapi import FastAPI, WebSocket
from starlette.websockets import WebSocketDisconnect
from contextlib import suppress

from csrd.realtime import (
    ConnectionContext,
    ConnectionManager,
    ConnectionNotFoundError,
    MessageEnvelope,
    MessageRouter,
)

app = FastAPI()
manager = ConnectionManager()
router = MessageRouter()


@app.websocket("/ws")
async def ws_endpoint(ws: WebSocket):
    await ws.accept()
    connection_id = ws.query_params.get("connection_id", "anon")
    context = ConnectionContext(connection_id=connection_id)
    await manager.connect(ws, context)

    try:
        while True:
            data = await ws.receive_json()
            envelope = MessageEnvelope.model_validate(data)
            await router.dispatch(connection_id, envelope, manager)
    except WebSocketDisconnect:
        pass
    finally:
        with suppress(ConnectionNotFoundError):
            await manager.disconnect(connection_id)

Optional context adapter pattern

csrd-realtime stays standalone, but supports optional context binding around each dispatch via RealtimeContextAdapter.

Use this to bridge into csrd-context/logging contextvars without adding a hard dependency from realtime core.

from csrd.realtime import MessageRouter, RealtimeContextAdapter


def bind_ws_context(dispatch_context):
    # Example: set your contextvars and return reset token(s)
    token = some_contextvar.set(
        {
            "connection_id": dispatch_context.connection_id,
            "tenant_id": dispatch_context.connection.tenant_id,
            "event_type": dispatch_context.event_type,
            "trace_id": dispatch_context.trace_id,
        }
    )
    return token


def reset_ws_context(token):
    some_contextvar.reset(token)


router = MessageRouter(
    context_adapter=RealtimeContextAdapter(bind=bind_ws_context, reset=reset_ws_context)
)

Operational defaults (current)

• Backpressure model: direct send_json per message (no per-connection queue in v0.1.x).
• Failure behavior: if a send fails, ConnectionManager evicts that connection (best-effort fan-out).
• Timeout behavior: manager applies bounded send timeout by default (send_timeout_seconds=5.0).
• Router behavior: handler exceptions are surfaced to caller by default; set MessageRouter(raise_handler_exceptions=False) for log-and-continue mode.

Tenant and room naming guidance

Room names are consumer-defined strings. Use a namespaced format to avoid cross-tenant collisions.

Recommended patterns:

• tenant:{tenant_id}:session:{session_id} for gameplay sessions
• tenant:{tenant_id}:party:{party_id} for party-level channels
• tenant:{tenant_id}:system for scoped system broadcasts

Guidance:

• Always resolve tenant_id server-side from auth context.
• Never trust client-provided room names without authorization checks.
• Keep room IDs stable and deterministic so reconnect/join logic is predictable.