tl/scenes/consumers.py

import json
import uuid

from channels.generic.websocket import AsyncWebsocketConsumer

# Who is connected, per project: {group: {cid: {...presence...}}}. Purely a
# cache of what has already been relayed, so a joiner can be handed the room in
# one message instead of asking everyone in it to answer (which costs one
# broadcast per member, i.e. N^2 messages for a room of N — at 100 that loses
# most of the roster to the channel layer's queue limits). Nothing about a party
# is decided here; the browsers still work that out from the roster (tl.party).
#
# Process-local, like the in-memory channel layer this runs on. If that ever
# moves to Redis for multiple workers, this has to move with it.
ROOMS = {}


class SceneConsumer(AsyncWebsocketConsumer):
    """One connection per open project. Two things ride it.

    Scene deltas: relayed from what the server broadcasts on a save (see
    scenes.views.scene). Read-only — edits still go through the PUT endpoint,
    which is the broadcast source.

    Presence: who's viewing, which ad-hoc party they're in, and where that party
    is looking. Peers gossip it between themselves; all this does is hand out a
    connection id, stamp the sender's server-side identity onto every message —
    so nobody can post as somebody else — and relay.
    """

    RELAYED = ("state", "nav")

    @property
    def room(self):
        return ROOMS.setdefault(self.group, {})

    async def connect(self):
        self.pk = self.scope["url_route"]["kwargs"]["pk"]
        self.group = f"project_{self.pk}"
        self.cid = uuid.uuid4().hex[:12]
        user = self.scope.get("user")
        self.username = user.get_username() if user and user.is_authenticated else None
        await self.channel_layer.group_add(self.group, self.channel_name)
        await self.accept()

        me = {"cid": self.cid, "user": self.username, "party": None, "joinable": True}
        others = list(self.room.values())
        self.room[self.cid] = me
        await self.send(text_data=json.dumps({"kind": "welcome", **me}))
        await self.send(text_data=json.dumps({"kind": "roster", "peers": others}))
        await self._relay({"kind": "join"})

    async def disconnect(self, code):
        if hasattr(self, "cid"):
            self.room.pop(self.cid, None)
            if not self.room:
                ROOMS.pop(self.group, None)
            await self._relay({"kind": "leave"})
            await self.channel_layer.group_discard(self.group, self.channel_name)

    async def receive(self, text_data=None, bytes_data=None):
        try:
            msg = json.loads(text_data or "{}")
        except ValueError:
            return
        if not isinstance(msg, dict) or msg.get("kind") not in self.RELAYED:
            return
        if msg.get("kind") == "state" and self.cid in self.room:
            self.room[self.cid].update(party=msg.get("party"),
                                       joinable=bool(msg.get("joinable")))
        await self._relay(msg)

    async def _relay(self, msg):
        await self.channel_layer.group_send(
            self.group,
            {"type": "peer.msg", "msg": {**msg, "cid": self.cid, "user": self.username}})

    # broadcast handler: {type: "peer.msg", msg: {...}} — presence gossip
    async def peer_msg(self, event):
        await self.send(text_data=json.dumps(event["msg"]))

    # broadcast handler: {type: "scene.delta", delta: {...}}
    async def scene_delta(self, event):
        await self.send(text_data=json.dumps(event["delta"]))