Django Channels

Django==6.1.2
channels==4.3.2
daphne==4.2.3
channels-redis==4.3.0
redis==7.4.1

FROM python:3.13.17-slim-trixie
ENV PYTHONUNBUFFERED=1 PYTHONDONTWRITEBYTECODE=1
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["daphne", "-b", "0.0.0.0", "-p", "8000", "config.asgi:application"]

# docker compose -p chatterbox up --build
services:
  redis:
    image: redis:8.8.3-alpine
    ports:
      - "127.0.0.1:${REDIS_PORT:-6379}:6379"
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 2s
      timeout: 2s
      retries: 15

  web:
    build: .
    command: sh -c "python manage.py migrate --noinput && daphne -b 0.0.0.0 -p 8000 --websocket-max-message-size 65536 config.asgi:application"
    environment:
      REDIS_URL: redis://redis:6379/0
      DJANGO_SECRET_KEY: change-me
      ALLOWED_HOSTS: localhost,127.0.0.1
      CHATTERBOX_ORIGINS: http://localhost:8000
      SQLITE_PATH: /data/db.sqlite3
    volumes:
      - data:/data
    ports:
      - "127.0.0.1:${WEB_PORT:-8000}:8000"
    depends_on:
      redis:
        condition: service_healthy

  worker:
    build: .
    command: python manage.py runworker --layer game arena
    environment:
      REDIS_URL: redis://redis:6379/0
      DJANGO_SECRET_KEY: change-me
      SQLITE_PATH: /data/db.sqlite3
    volumes:
      - data:/data
    depends_on:
      redis:
        condition: service_healthy

volumes:
  data:

chatterbox/
    compose.yaml  Dockerfile  requirements.txt  manage.py
    config/   settings.py  asgi.py  urls.py
    chat/
        models.py          # Room, Membership, Event, Processed
        protocol.py        # envelope helpers, ChatError, validation
        store.py           # synchronous ORM half (run in threads)
        presence.py        # Redis: presence, tickets, typing throttle, sessions
        redis_client.py    # one redis-py client per event loop
        consumers.py       # ChatConsumer
        arena.py           # ArenaWorker and ArenaConsumer
        security.py        # OptionalOriginValidator
        routing.py  views.py  management/commands/mint_ticket.py
        tests/test_chat.py

# config/settings.py (excerpt)
CHATTERBOX_ORIGINS = os.environ.get("CHATTERBOX_ORIGINS", "http://localhost:8000").split(",")
REDIS_URL = os.environ.get("REDIS_URL")

DATABASES = {
    "default": {
        "ENGINE": "django.db.backends.sqlite3",
        "NAME": os.environ.get("SQLITE_PATH", BASE_DIR / "db.sqlite3"),
        # BEGIN IMMEDIATE: take the write lock at the start of atomic(), so the
        # per-room seq counter cannot be read stale (Django 5.1+).
        "OPTIONS": {"transaction_mode": "IMMEDIATE", "timeout": 20},
    }
}

if REDIS_URL:
    CHANNEL_LAYERS = {
        # chat: durable-ish events, a minute of grace for slow consumers
        "default": {
            "BACKEND": "channels_redis.core.RedisChannelLayer",
            "CONFIG": {
                "hosts": [REDIS_URL],
                "prefix": "chatterbox",
                "expiry": 60,
                "capacity": 200,
            },
        },
        # game: snapshots are worthless after a second, so expire fast and
        # keep the queues short; the worker's own queue gets more room
        "game": {
            "BACKEND": "channels_redis.core.RedisChannelLayer",
            "CONFIG": {
                "hosts": [REDIS_URL],
                "prefix": "arena",
                "expiry": 1,
                "capacity": 10,
                "channel_capacity": {"arena": 500},
            },
        },
    }
else:
    CHANNEL_LAYERS = {
        "default": {"BACKEND": "channels.layers.InMemoryChannelLayer"},
        "game": {"BACKEND": "channels.layers.InMemoryChannelLayer"},
    }

# config/asgi.py
import os

from django.core.asgi import get_asgi_application

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "config.settings")
django_asgi_app = get_asgi_application()  # django.setup() before model imports

from channels.auth import AuthMiddlewareStack  # noqa: E402
from channels.routing import ChannelNameRouter, ProtocolTypeRouter, URLRouter  # noqa: E402
from django.conf import settings  # noqa: E402

from chat import routing  # noqa: E402
from chat.arena import ArenaWorker  # noqa: E402
from chat.security import OptionalOriginValidator  # noqa: E402

application = ProtocolTypeRouter(
    {
        "http": django_asgi_app,
        "websocket": OptionalOriginValidator(
            AuthMiddlewareStack(URLRouter(routing.websocket_urlpatterns)),
            settings.CHATTERBOX_ORIGINS,
        ),
        "channel": ChannelNameRouter({"arena": ArenaWorker.as_asgi()}),
    }
)

# chat/security.py
from channels.security.websocket import OriginValidator


class OptionalOriginValidator(OriginValidator):
    """Allow-list the Origin header, but let clients that send none through.

    Browsers always send Origin, and a hostile page cannot remove it, so the
    cross-site hijacking check still holds. Non-browser clients (scripts,
    mobile apps) send no Origin and authenticate with a ticket instead.
    """

    def valid_origin(self, parsed_origin):
        if parsed_origin is None:
            return True
        return self.validate_origin(parsed_origin)

# chat/models.py
from django.conf import settings
from django.db import models


class Room(models.Model):
    name = models.CharField(max_length=64, primary_key=True)
    last_seq = models.PositiveIntegerField(default=0)  # per-room counter


class Membership(models.Model):
    user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE)
    room = models.ForeignKey(Room, on_delete=models.CASCADE)

    class Meta:
        constraints = [
            models.UniqueConstraint(fields=["user", "room"], name="one_membership")
        ]


class Event(models.Model):
    """One row per seq: a message, an edit of a message or a reaction."""

    class Kind(models.TextChoices):
        MESSAGE = "message"
        EDIT = "edit"
        REACT = "react"

    room = models.ForeignKey(Room, on_delete=models.CASCADE)
    seq = models.PositiveIntegerField()
    kind = models.CharField(max_length=8, choices=Kind)
    ref = models.CharField(max_length=36)  # message id, or the target of edit/react
    user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE)
    data = models.JSONField()
    ts = models.BigIntegerField()  # server time, epoch milliseconds

    class Meta:
        ordering = ["seq"]
        constraints = [
            models.UniqueConstraint(fields=["room", "seq"], name="seq_unique_per_room"),
            models.UniqueConstraint(
                fields=["ref"],
                condition=models.Q(kind="message"),
                name="message_id_unique",
            ),
        ]


class Processed(models.Model):
    """Idempotency table: the first ack of every (user, envelope id)."""

    user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE)
    env_id = models.CharField(max_length=36)
    ack = models.JSONField()
    created = models.DateTimeField(auto_now_add=True)

    class Meta:
        constraints = [
            models.UniqueConstraint(fields=["user", "env_id"], name="one_ack_per_id")
        ]

# chat/consumers.py
import asyncio
import logging
import secrets
import time
from collections import deque

from channels.db import database_sync_to_async
from channels.generic.websocket import AsyncJsonWebsocketConsumer
from django.contrib.auth import get_user_model
from django.db import IntegrityError

from . import presence, store
from .protocol import (
    HEARTBEAT_MS, ID_RE, MAX_FRAME_BYTES, ROOM_RE, SUBPROTOCOL, ChatError,
    ack_envelope, envelope, error_envelope, parse_envelope,
)  # fmt: skip

log = logging.getLogger("chatterbox")
GRACE_SECONDS = 15  # offline is announced this long after the last socket closes
_background = set()  # strong references: the loop only keeps weak ones


def room_group(room):
    return f"room-{room}"  # [a-z0-9-] is valid in a group name


class TokenBucket:
    def __init__(self, rate=20.0, capacity=40.0):
        self.rate, self.capacity = rate, capacity
        self.tokens, self.stamp = capacity, time.monotonic()

    def take(self):
        now = time.monotonic()
        self.tokens = min(self.capacity, self.tokens + (now - self.stamp) * self.rate)
        self.stamp = now
        if self.tokens >= 1:
            self.tokens -= 1
            return True
        return False


async def announce_offline_later(user_id, name, rooms, layer):
    await asyncio.sleep(GRACE_SECONDS)
    if await presence.status_of(user_id) != "offline":
        return  # reconnected during the grace period
    for room in rooms:
        payload = {"userId": user_id, "name": name, "status": "offline"}
        await layer.group_send(
            room_group(room),
            {"type": "chat.envelope", "envelope": envelope("presence", payload, room),
             "exclude_user": user_id},
        )  # fmt: skip


class ChatConsumer(AsyncJsonWebsocketConsumer):
    user = None
    uid = None
    session_id = None
    watchdog_task = None
    rooms = ()
    closing = False  # set once we sent a close: later frames are ignored

    # ---- connection lifecycle -------------------------------------------
    async def connect(self):
        offered = self.scope.get("subprotocols", [])
        if offered and SUBPROTOCOL not in offered:
            await self.close()  # handshake refused: HTTP 403
            return
        self.rooms = set()
        self.bucket = TokenBucket()
        self.limit_hits = deque()
        self.last_rx = time.monotonic()
        self.said_hello = asyncio.Event()
        await self.accept(SUBPROTOCOL if offered else None)
        self.watchdog_task = asyncio.create_task(self._watchdog())

    async def _watchdog(self):
        try:
            await asyncio.wait_for(self.said_hello.wait(), 5)
        except asyncio.TimeoutError:
            await self._close(4408)  # no hello within 5 s
            return
        while True:
            await asyncio.sleep(HEARTBEAT_MS / 1000)
            if time.monotonic() - self.last_rx > 2 * HEARTBEAT_MS / 1000:
                await self._close(4408)  # two heartbeats of silence
                return

    async def disconnect(self, code):
        if self.watchdog_task:
            self.watchdog_task.cancel()
        if self.user is None:
            return
        for room in self.rooms:
            await self.channel_layer.group_discard(room_group(room), self.channel_name)
        if await presence.drop(self.uid, self.channel_name) == 0:
            await presence.save_session(self.session_id, self.uid, presence.SESSION_TTL_S)
            task = asyncio.create_task(
                announce_offline_later(
                    self.uid, self.user.get_username(), list(self.rooms), self.channel_layer
                )
            )
            _background.add(task)
            task.add_done_callback(_background.discard)

    # ---- inbound frames ---------------------------------------------------
    async def _close(self, code):
        self.closing = True
        await self.close(code=code)

    async def receive(self, text_data=None, bytes_data=None, **kwargs):
        if self.closing:
            return  # frames queued behind the one that got us closed
        self.last_rx = time.monotonic()
        if bytes_data is not None or text_data is None:
            await self._close(4400)  # text frames only
            return
        if len(text_data.encode()) > MAX_FRAME_BYTES:
            await self._send_error(None, None, ChatError("too_large", "frame too large", close=4413))
            return
        try:
            frame = await self.decode_json(text_data)
        except ValueError:
            await self._send_error(None, None, ChatError("bad_request", "not JSON"))
            return
        await self.receive_json(frame)

    async def receive_json(self, frame, **kwargs):
        ref = room = None
        if isinstance(frame, dict):  # echo ids back only when they look valid
            ref = frame["id"] if isinstance(frame.get("id"), str) and ID_RE.match(frame["id"]) else None
            room = frame["room"] if isinstance(frame.get("room"), str) and ROOM_RE.match(frame["room"]) else None
        try:
            self._rate_limit()
            await self._dispatch(parse_envelope(frame))
        except ChatError as err:
            await self._send_error(ref, room, err)
        except IntegrityError:  # lost a race on a unique constraint
            await self._send_error(ref, room, ChatError("conflict", "id already used"))
        except Exception:
            log.exception("unhandled error in chat consumer")
            await self._send_error(ref, room, ChatError("internal", "internal error"))

    def _rate_limit(self):
        if self.bucket.take():
            return
        now = time.monotonic()
        self.limit_hits.append(now)
        while self.limit_hits and now - self.limit_hits[0] > 10:
            self.limit_hits.popleft()
        close = 4429 if len(self.limit_hits) >= 10 else None
        raise ChatError("rate_limited", "slow down", close=close)

    async def _send_error(self, ref, room, err):
        await self.send_json(error_envelope(ref, room, err))
        if err.close:
            await self._close(err.close)

    async def _dispatch(self, env):
        if env["type"] == "hello":
            return await self._on_hello(env)
        if self.user is None:
            raise ChatError("unauthorized", "send hello first", close=4401)
        await getattr(self, f"_on_{env['type']}")(env)

    # ---- helpers ----------------------------------------------------------
    @staticmethod
    def _room(env):
        if env["room"] is None:
            raise ChatError("bad_request", "room is required")
        return env["room"]

    async def _db(self, fn, *args):
        return await database_sync_to_async(fn)(*args)

    async def _fanout(self, room, env, exclude_user=None):
        await self.channel_layer.group_send(
            room_group(room),
            {"type": "chat.envelope", "envelope": env, "exclude_user": exclude_user},
        )

    async def _attach(self, room):
        """Make sure this socket receives the room's events (idempotent)."""
        if room not in self.rooms:
            self.rooms.add(room)
            await self.channel_layer.group_add(room_group(room), self.channel_name)

    async def _acked(self, env, handler):
        out = await self._db(store.run_once, self.user, env["id"], handler)
        return out

    async def _finish(self, env, out):
        await self.send_json(ack_envelope(env["id"], env["room"], out.seq, out.dup, out.result))
        for b in out.broadcast:
            await self._fanout(b.room, b.envelope, b.exclude_user)

    async def _presence_env(self, room, status):
        payload = {"userId": self.uid, "name": self.user.get_username(), "status": status}
        return envelope("presence", payload, room)

    async def chat_envelope(self, event):
        """Group handler for 'chat.envelope': one frame per subscribed socket."""
        if event["exclude_user"] == self.uid:
            return
        await self.send_json(event["envelope"])

    # ---- message types ----------------------------------------------------
    async def _on_hello(self, env):
        if self.user is not None:
            raise ChatError("conflict", "hello was already completed")
        payload = env["payload"]
        user = self.scope.get("user")
        if user is None or not user.is_authenticated:  # no cookie: need a ticket
            uid = await presence.redeem_ticket(payload.get("token"))
            user = await get_user_model().objects.filter(pk=uid, is_active=True).afirst() if uid else None
            if user is None:
                raise ChatError("unauthorized", "invalid or expired ticket", close=4401)
        self.user, self.uid = user, str(user.pk)
        resume = payload.get("resume") if isinstance(payload.get("resume"), dict) else {}
        resumed = bool(resume) and await presence.session_owner(resume.get("sessionId")) == self.uid
        self.session_id = resume["sessionId"] if resumed else secrets.token_hex(16)
        before = await presence.status_of(self.uid)
        rooms = await self._db(store.rooms_of, user)
        for room in rooms:
            await self._attach(room)
        await presence.beat(self.uid, self.channel_name, "online" if before == "offline" else before)
        await presence.save_session(self.session_id, self.uid)
        self.said_hello.set()
        await self.send_json(envelope("hello", {
            "userId": self.uid, "name": user.get_username(), "sessionId": self.session_id,
            "heartbeatMs": HEARTBEAT_MS, "maxFrameBytes": MAX_FRAME_BYTES,
            "resumed": resumed, "rooms": sorted(rooms), "ref": env["id"],
        }))  # fmt: skip
        if resumed and isinstance(resume.get("rooms"), dict):
            for room, last in resume["rooms"].items():
                if room in self.rooms and isinstance(last, int):
                    for item in await self._db(store.replay, room, last):
                        await self.send_json(item)
        if before == "offline":
            for room in self.rooms:
                await self._fanout(room, await self._presence_env(room, "online"), self.uid)

    async def _on_join(self, env):
        room = self._room(env)
        out = await self._acked(env, lambda: store.join(self.user, room, env["payload"]))
        await self._attach(room)
        for member in out.result["members"]:  # live data, so never cached
            member["status"] = await presence.status_of(member["userId"])
        await self._finish(env, out)
        for item in out.replay:
            await self.send_json(item)
        if out.joined:
            status = await presence.status_of(self.uid)
            await self._fanout(room, await self._presence_env(room, status), self.uid)

    async def _on_leave(self, env):
        room = self._room(env)
        out = await self._acked(env, lambda: store.leave(self.user, room))
        self.rooms.discard(room)
        await self.channel_layer.group_discard(room_group(room), self.channel_name)
        await self._finish(env, out)

    async def _on_message(self, env):
        room = self._room(env)
        out = await self._acked(env, lambda: store.post_message(self.user, room, env["id"], env["payload"]))
        await self._attach(room)
        await self._finish(env, out)

    async def _on_edit(self, env):
        room = self._room(env)
        out = await self._acked(env, lambda: store.edit(self.user, room, env["payload"]))
        await self._attach(room)
        await self._finish(env, out)

    async def _on_react(self, env):
        room = self._room(env)
        out = await self._acked(env, lambda: store.react(self.user, room, env["payload"]))
        await self._attach(room)
        await self._finish(env, out)

    async def _on_history(self, env):
        room = self._room(env)
        out = await self._db(store.history, self.user, room, env["payload"])
        await self._finish(env, out)

    async def _on_presence(self, env):
        status = env["payload"].get("status")
        if status not in ("online", "away"):
            raise ChatError("bad_request", "status must be online or away")
        before = await presence.status_of(self.uid)
        await presence.beat(self.uid, self.channel_name, status)  # also the heartbeat
        await self.send_json(ack_envelope(env["id"], env["room"]))
        if before != status:
            for room in self.rooms:
                await self._fanout(room, await self._presence_env(room, status), self.uid)

    async def _on_typing(self, env):
        room, state = self._room(env), env["payload"].get("state")
        if state not in ("start", "stop"):
            raise ChatError("bad_request", "state must be start or stop")
        if room not in self.rooms:
            raise ChatError("forbidden", "join the room first")
        if state == "start" and not await presence.allow_typing(self.uid, room):
            return  # throttled: dropped silently, fire-and-forget
        payload = {"userId": self.uid, "state": state}
        await self._fanout(room, envelope("typing", payload, room), self.uid)

    async def _on_receipt(self, env):
        room, p = self._room(env), env["payload"]
        if p.get("kind") not in ("delivered", "read") or not isinstance(p.get("upTo"), int):
            raise ChatError("bad_request", "kind and upTo are required")
        if room not in self.rooms:
            raise ChatError("forbidden", "join the room first")
        if p["upTo"] > await self._db(store.last_seq, room):
            raise ChatError("bad_request", "upTo is beyond the last seq")
        payload = {"userId": self.uid, "kind": p["kind"], "upTo": p["upTo"]}
        await self._fanout(room, envelope("receipt", payload, room), self.uid)

# chat/store.py (excerpt)
@dataclass
class Broadcast:
    room: str
    envelope: dict
    exclude_user: str | None = None  # "others only" types skip the sender


@dataclass
class Outcome:
    seq: int | None
    result: dict
    dup: bool = False
    broadcast: list = field(default_factory=list)
    replay: list = field(default_factory=list)  # for this connection only
    joined: bool = False


def run_once(user, env_id, handler):
    """Run handler() at most once per (user, envelope id), atomically.

    The ack is stored in the same transaction as the change it acknowledges,
    so a crash leaves either both or neither. A ChatError rolls everything
    back, which is why errors are never cached.
    """
    with transaction.atomic():
        done = Processed.objects.filter(user=user, env_id=env_id).first()
        if done:
            return Outcome(done.ack["seq"], done.ack["result"], dup=True)
        outcome = handler()
        Processed.objects.create(
            user=user, env_id=env_id, ack={"seq": outcome.seq, "result": outcome.result}
        )
    return outcome


def _next_seq(room):
    # UPDATE first: the write lock (BEGIN IMMEDIATE / a row lock) is held
    # before the value is read back, so two writers never see the same seq.
    Room.objects.filter(pk=room).update(last_seq=F("last_seq") + 1)
    return Room.objects.values_list("last_seq", flat=True).get(pk=room)


def post_message(user, room, env_id, payload):
    _require_member(user, room)
    text = clean_text(payload.get("text"))
    reply_to = payload.get("replyTo")
    attachments = payload.get("attachments", [])
    if not isinstance(attachments, list) or len(attachments) > 5:
        raise ChatError("bad_request", "at most 5 attachments")
    for a in attachments:
        if not (isinstance(a, dict) and str(a.get("url", "")).startswith("https://")):
            raise ChatError("bad_request", "attachments need an https url")
    if reply_to is not None and not Event.objects.filter(
        room_id=room, kind=Event.Kind.MESSAGE, ref=reply_to
    ).exists():
        raise ChatError("not_found", "replyTo is not a message of this room")
    if Event.objects.filter(kind=Event.Kind.MESSAGE, ref=env_id).exists():
        raise ChatError("conflict", "message id already used")
    seq = _next_seq(room)
    event = Event.objects.create(
        room_id=room, seq=seq, kind=Event.Kind.MESSAGE, ref=env_id, user=user,
        ts=now_ms(), data={"text": text, "replyTo": reply_to, "attachments": attachments},
    )  # fmt: skip
    return Outcome(seq, {}, broadcast=[Broadcast(room, _message_env(event))])


def history(user, room, payload):
    _require_member(user, room)
    before, limit = payload.get("before"), payload.get("limit", 50)
    if not isinstance(limit, int) or not 1 <= limit <= 100:
        raise ChatError("bad_request", "limit must be 1..100")
    rows = Event.objects.filter(room_id=room, kind=Event.Kind.MESSAGE)
    if before is not None:
        rows = rows.filter(seq__lt=before)
    rows = list(rows.order_by("-seq").select_related("user")[: limit + 1])
    page, size = [], 0
    for event in rows[:limit]:
        size += len(json.dumps(_message_env(event)))
        if page and size > MAX_PAGE_BYTES:
            break
        page.append(event)
    edits = {}  # apply the latest edit of every message in the page
    for e in Event.objects.filter(
        room_id=room, kind=Event.Kind.EDIT, ref__in=[m.ref for m in page]
    ).order_by("seq"):
        edits[e.ref] = e.data["text"]
    messages = [
        _message_env(m, edits.get(m.ref), m.ref in edits) for m in reversed(page)
    ]
    has_more = len(rows) > len(page)
    nxt = page[-1].seq if has_more else None
    return Outcome(None, {"messages": messages, "hasMore": has_more, "next": nxt})

# chat/presence.py
"""Presence in Redis. Every key expires on its own, so a crashed worker cannot
leave a user online forever. Nothing here ever calls KEYS or SCAN."""

from .protocol import HEARTBEAT_MS
from .redis_client import get_redis

TTL_MS = 2 * HEARTBEAT_MS  # two missed heartbeats = gone
TICKET_TTL_S = 30
SESSION_TTL_S = 120


async def _now_ms(r):
    sec, usec = await r.time()  # Redis clock: no skew between app servers
    return sec * 1000 + usec // 1000


async def beat(user_id: str, channel: str, status: str) -> None:
    """Register or refresh one connection of a user."""
    r = get_redis()
    now = await _now_ms(r)
    async with r.pipeline(transaction=True) as p:
        p.zadd(f"cb:conns:{user_id}", {channel: now + TTL_MS})
        p.pexpire(f"cb:conns:{user_id}", TTL_MS * 2)
        p.set(f"cb:status:{user_id}", status, px=TTL_MS * 2)
        await p.execute()


async def drop(user_id: str, channel: str) -> int:
    """Remove one connection; return how many live connections remain."""
    r = get_redis()
    await r.zrem(f"cb:conns:{user_id}", channel)
    return await live_connections(user_id)


async def live_connections(user_id: str) -> int:
    r = get_redis()
    now = await _now_ms(r)
    key = f"cb:conns:{user_id}"
    await r.zremrangebyscore(key, "-inf", now)  # prune crashed connections
    return await r.zcard(key)


async def status_of(user_id: str) -> str:
    if await live_connections(user_id) == 0:
        return "offline"
    return await get_redis().get(f"cb:status:{user_id}") or "online"


async def allow_typing(user_id: str, room: str) -> bool:
    """At most one typing 'start' per user and room every 3 seconds."""
    return bool(await get_redis().set(f"cb:typing:{user_id}:{room}", 1, nx=True, px=3000))


async def issue_ticket(user_id: str) -> str:
    import secrets

    token = secrets.token_urlsafe(24)
    await get_redis().set(f"cb:ticket:{token}", user_id, ex=TICKET_TTL_S)
    return token


async def redeem_ticket(token) -> str | None:
    if not isinstance(token, str) or not 8 <= len(token) <= 64:
        return None
    return await get_redis().getdel(f"cb:ticket:{token}")  # one use only


async def save_session(session_id: str, user_id: str, ttl: int = 86400) -> None:
    await get_redis().set(f"cb:session:{session_id}", user_id, ex=ttl)


async def session_owner(session_id) -> str | None:
    if not isinstance(session_id, str):
        return None
    return await get_redis().get(f"cb:session:{session_id}")

# chat/redis_client.py
import asyncio
import weakref

from django.conf import settings
from redis.asyncio import Redis

_clients = weakref.WeakKeyDictionary()


def get_redis() -> Redis:
    """One client per running event loop (each test gets its own loop)."""
    loop = asyncio.get_running_loop()
    client = _clients.get(loop)
    if client is None:
        client = _clients[loop] = Redis.from_url(settings.REDIS_URL, decode_responses=True)
    return client

# chat/arena.py
"""Pixel Arena: an authoritative 20 Hz game loop in a channel-layer worker."""

import asyncio
import time

from channels.consumer import AsyncConsumer
from channels.generic.websocket import AsyncJsonWebsocketConsumer

from . import presence

TICK_HZ = 20
SPEED = 120.0  # pixels per second
WIDTH, HEIGHT = 800, 600
STATE_GROUP = "arena-state"


class ArenaWorker(AsyncConsumer):
    """Runs under `manage.py runworker arena`; receives events sent to "arena".

    One process owns the world, so run exactly one of these per arena.
    """

    channel_layer_alias = "game"

    players = {}  # player id -> {"x", "y", "dx", "dy", "ack"}; class-level: one world
    tick = 0
    loop_task = None

    async def arena_join(self, event):
        self.players.setdefault(event["player"], {"x": 400.0, "y": 300.0, "dx": 0, "dy": 0, "ack": 0})
        self._ensure_loop()

    async def arena_leave(self, event):
        self.players.pop(event["player"], None)

    async def arena_input(self, event):
        p = self.players.get(event["player"])
        if p is not None and event["seq"] > p["ack"]:
            p["dx"], p["dy"], p["ack"] = event["dx"], event["dy"], event["seq"]

    def _ensure_loop(self):
        if ArenaWorker.loop_task is None or ArenaWorker.loop_task.done():
            ArenaWorker.loop_task = asyncio.create_task(self._run())

    async def _run(self):
        dt = 1 / TICK_HZ
        due = time.monotonic()
        while self.players:
            for p in self.players.values():  # simulate with the latest input
                p["x"] = min(WIDTH, max(0.0, p["x"] + p["dx"] * SPEED * dt))
                p["y"] = min(HEIGHT, max(0.0, p["y"] + p["dy"] * SPEED * dt))
            ArenaWorker.tick += 1
            snapshot = {pid: [round(p["x"], 1), round(p["y"], 1), p["ack"]] for pid, p in self.players.items()}
            await self.channel_layer.group_send(
                STATE_GROUP,
                {"type": "arena.snapshot", "tick": self.tick, "players": snapshot},
            )
            due += dt  # fixed-step schedule: no drift when a tick runs long
            await asyncio.sleep(max(0.0, due - time.monotonic()))


class ArenaConsumer(AsyncJsonWebsocketConsumer):
    """The player's socket: forwards inputs to the worker, relays snapshots."""

    channel_layer_alias = "game"
    player = None
    deadline = None

    async def connect(self):
        await self.accept()
        user = self.scope["user"]
        if user.is_authenticated:  # browser with a session cookie
            await self._enter(str(user.pk))
        else:  # script or app: first frame must be {"type": "hello", "token": ...}
            self.deadline = asyncio.get_running_loop().call_later(
                5, lambda: asyncio.create_task(self.close(code=4401))
            )

    async def _enter(self, player):
        if self.deadline:
            self.deadline.cancel()
        self.player = player
        await self.channel_layer.group_add(STATE_GROUP, self.channel_name)
        await self.channel_layer.send("arena", {"type": "arena.join", "player": player})
        await self.send_json({"type": "welcome", "player": player, "tickHz": TICK_HZ})

    async def disconnect(self, code):
        if self.deadline:
            self.deadline.cancel()
        if self.player is not None:
            await self.channel_layer.group_discard(STATE_GROUP, self.channel_name)
            await self.channel_layer.send("arena", {"type": "arena.leave", "player": self.player})

    async def receive_json(self, content, **kwargs):
        if self.player is None:
            player = await presence.redeem_ticket(content.get("token") if isinstance(content, dict) else None)
            if player is None:
                await self.close(code=4401)
            else:
                await self._enter(player)
            return
        try:
            seq, dx, dy = int(content["seq"]), int(content["dx"]), int(content["dy"])
        except (KeyError, TypeError, ValueError):
            return  # ignore malformed input, never crash the loop
        await self.channel_layer.send(
            "arena",
            {"type": "arena.input", "player": self.player, "seq": seq,
             "dx": max(-1, min(1, dx)), "dy": max(-1, min(1, dy))},
        )  # fmt: skip

    async def arena_snapshot(self, event):
        await self.send_json({"type": "snapshot", "tick": event["tick"], "players": event["players"]})

docker compose -p chatterbox up -d --build      # redis, web (daphne) and worker
node client.mjs                                 # 36 checks, under a minute
docker compose -p chatterbox down -v            # stop and delete the volumes

// client.mjs (excerpt)
// Conformance run for the Django Chatterbox server (Node 24, global WebSocket).
import { execFileSync } from "node:child_process";
import { setTimeout as sleep } from "node:timers/promises";

const PORT = process.env.WEB_PORT ?? "8000";
const URL = `ws://127.0.0.1:${PORT}/ws`;
const PROJECT = process.env.COMPOSE_PROJECT ?? "chatterbox";
let failures = 0;
const check = (ok, label) => { console.log(`${ok ? "PASS" : "FAIL"}  ${label}`); if (!ok) failures++; };

const alphabet = "0123456789ABCDEFGHJKMNPQRSTVWXYZ";
const ulid = () => Array.from({ length: 26 }, () => alphabet[Math.floor(Math.random() * 32)]).join("");

function ticket(name) {
  const out = execFileSync("docker", ["compose", "-p", PROJECT, "exec", "-T", "web", "python", "manage.py", "mint_ticket", name]).toString().trim();
  const [id, token] = out.split(" ");
  return { id, token };
}

class Client {
  frames = []; closed = null;
  constructor(path = URL, protocol = "chatterbox.v1") {
    this.ws = protocol ? new WebSocket(path, protocol) : new WebSocket(path);
    this.opened = new Promise((res, rej) => { this.ws.onopen = res; this.ws.onerror = () => rej(new Error("connect failed")); });
    this.ws.onmessage = (e) => this.frames.push(JSON.parse(e.data));
    this.ws.onclose = (e) => { this.closed = e.code; };
  }
  send(type, payload = {}, room = null, id = ulid()) {
    this.ws.send(JSON.stringify({ v: 1, type, id, room, ts: 0, payload })); return id;
  }
  async next(pred, ms = 2000) {
    const end = Date.now() + ms;
    while (Date.now() < end) {
      const i = this.frames.findIndex(pred);
      if (i >= 0) return this.frames.splice(i, 1)[0];
      await sleep(10);
    }
    return null;
  }
  ack = (id, ms) => this.next((f) => f.type === "ack" && f.payload.ref === id, ms);
  error = (id, ms) => this.next((f) => f.type === "error" && f.payload.ref === id, ms);
  async closedWith(ms = 2000) { const end = Date.now() + ms; while (this.closed === null && Date.now() < end) await sleep(20); return this.closed; }
}

async function login(name) {
  const t = ticket(name);
  const c = new Client(); await c.opened;
  c.send("hello", { token: t.token, client: "node/24" });
  const hello = await c.next((f) => f.type === "hello");
  return { c, hello, t };
}

// ... sections 1-4: handshake errors, hello, ticket replay, join and presence ...

// 5. message, broadcast, ack
const mid = ulid();
a.send("message", { text: "Hello, Chatterbox", replyTo: null, attachments: [] }, "general", mid);
ack = await a.ack(mid);
check(ack?.payload.ok && ack.payload.seq === 1 && ack.payload.dup === false, "message -> ack seq 1, dup false");
const toBob = await b.next((f) => f.type === "message" && f.id === mid);
const toAlice = await a.next((f) => f.type === "message" && f.id === mid);
check(toBob?.payload.seq === 1 && toBob.payload.from.name === "alice" && toAlice, "broadcast reaches bob and the sender, id = message id");
// 6. duplicate
a.send("message", { text: "Hello, Chatterbox", replyTo: null, attachments: [] }, "general", mid);
ack = await a.ack(mid);
check(ack?.payload.dup === true && ack.payload.seq === 1, "same id again -> ack dup:true, same seq");
check((await b.next((f) => f.type === "message", 400)) === null, "duplicate is not broadcast again");

// ... sections 7-15 ...

# chat/tests/test_chat.py
from unittest.mock import AsyncMock, patch

from channels.routing import URLRouter
from channels.testing import WebsocketCommunicator
from django.contrib.auth import get_user_model
from django.test import TransactionTestCase, override_settings

from chat import routing
from chat.protocol import SUBPROTOCOL

IN_MEMORY = {
    "default": {"BACKEND": "channels.layers.InMemoryChannelLayer"},
    "game": {"BACKEND": "channels.layers.InMemoryChannelLayer"},
}
ID = "01JAB3W7Q8Z4N5R6T7V8X9Y0Z{}"  # 26 characters once formatted


def frame(type_, payload=None, room=None, n="A"):
    return {"v": 1, "type": type_, "id": ID.format(n), "room": room, "ts": 0, "payload": payload or {}}


# presence and tickets use Redis (the compose service); the layer is in memory
@override_settings(CHANNEL_LAYERS=IN_MEMORY)
@patch("chat.consumers.announce_offline_later", new=AsyncMock())
class ChatConsumerTests(TransactionTestCase):
    def setUp(self):
        User = get_user_model()
        self.alice = User.objects.create_user("alice")
        self.bob = User.objects.create_user("bob")

    async def client_for(self, user):
        router = URLRouter(routing.websocket_urlpatterns)
        c = WebsocketCommunicator(router, "/ws", subprotocols=[SUBPROTOCOL])
        c.scope["user"] = user  # skip cookies: the stack has its own tests
        connected, subprotocol = await c.connect()
        assert connected and subprotocol == SUBPROTOCOL
        await c.send_json_to(frame("hello", n="H"))
        assert (await c.receive_json_from())["type"] == "hello"
        return c

    async def until(self, client, type_):
        """Next frame of the given type (presence frames can arrive in between)."""
        while (item := await client.receive_json_from())["type"] != type_:
            pass
        return item

    async def test_message_is_acked_broadcast_and_deduplicated(self):
        a, b = await self.client_for(self.alice), await self.client_for(self.bob)
        for c, n in ((a, "J"), (b, "K")):
            await c.send_json_to(frame("join", {"since": None}, "general", n))
            await self.until(c, "ack")
        await self.until(a, "presence")  # alice sees bob arrive

        msg = frame("message", {"text": "Hello, Chatterbox"}, "general", "M")
        await a.send_json_to(msg)
        ack = await self.until(a, "ack")
        self.assertEqual((ack["payload"]["seq"], ack["payload"]["dup"]), (1, False))
        self.assertEqual((await self.until(b, "message"))["payload"]["text"], "Hello, Chatterbox")

        await a.send_json_to(msg)  # same id again
        dup = await self.until(a, "ack")
        self.assertEqual((dup["payload"]["dup"], dup["payload"]["seq"]), (True, 1))
        self.assertTrue(await b.receive_nothing())  # nothing new for bob

        await a.disconnect()
        await b.disconnect()

    async def test_non_member_cannot_post(self):
        a = await self.client_for(self.alice)
        await a.send_json_to(frame("message", {"text": "hi"}, "secret"))
        reply = await a.receive_json_from()
        self.assertEqual((reply["type"], reply["payload"]["code"]), ("error", "forbidden"))
        await a.disconnect()

    async def test_frame_before_hello_closes_with_4401(self):
        router = URLRouter(routing.websocket_urlpatterns)
        c = WebsocketCommunicator(router, "/ws")
        c.scope["user"] = self.alice
        await c.connect()
        await c.send_json_to(frame("message", {"text": "hi"}, "general"))
        self.assertEqual((await c.receive_json_from())["payload"]["code"], "unauthorized")
        self.assertEqual((await c.receive_output())["code"], 4401)