feat: Add websocket channel

hyperfocus/channel.py

@@ -0,0 +1,107 @@
+from collections import defaultdict
+from typing import Optional, TypeAlias, TypedDict
+import json
+
+from datetime import datetime, timedelta
+
+from meta.sockets import Channel
+from modules.profiles.profiles.profiles import ProfilesRegistry
+
+from . import logger
+from .data import Hyperfocuser, HyperfocusData
+
+ISOTimestamp: TypeAlias = str
+
+
+class HyperfocusedPayload(TypedDict):
+    userid: str
+    user_name: str
+    started_at: ISOTimestamp
+    ends_at: ISOTimestamp
+
+
+async def prepare_hyperfocuser(profiler: ProfilesRegistry, hyperfocuser: Hyperfocuser):
+    profile = await profiler.get_profile(hyperfocuser.profileid)
+    assert profile is not None
+
+    return HyperfocusedPayload(
+        userid=str(hyperfocuser.profileid),
+        user_name=profile.nickname or "Unknown",
+        started_at=hyperfocuser.started_at.isoformat(),
+        ends_at=hyperfocuser.ends_at.isoformat(),
+    )
+
+
+class FocusChannel(Channel):
+    name = "HyperFocus"
+
+    def __init__(self, profiler: ProfilesRegistry, focusdata: HyperfocusData, **kwargs):
+        super().__init__(**kwargs)
+
+        self.profiler = profiler
+        self.focusdata = focusdata
+
+        # communityid -> listening websockets
+        self.communities = defaultdict(set)
+
+    async def on_connection(self, websocket, event):
+        if not (cidstr := event.get("community")):
+            raise ValueError("Hyperfocus browser missing communityid")
+        elif not cidstr.isdigit():
+            raise ValueError("Community id provided is not an integer")
+        cid = int(cidstr)
+        community = await self.profiler.get_community(cid)
+        if community is None:
+            raise ValueError("Unknown community provided")
+
+        await super().on_connection(websocket, event)
+        self.communities[cid].add(websocket)
+
+        focus_rows = await Hyperfocuser.fetch_where(started_in=cid)
+        await self.send_hyperfocus_put(cid, focus_rows)
+
+    async def del_connection(self, websocket):
+        for wss in self.communities.values():
+            wss.discard(websocket)
+        await super().del_connection(websocket)
+
+    async def send_hyperfocus_patch(
+        self, communityid: int, focuser: Hyperfocuser, websocket=None
+    ):
+        for ws in (websocket,) if websocket else self.communities[communityid]:
+            await self.send_event(
+                {
+                    "type": "DO",
+                    "method": "patchFocus",
+                    "args": await prepare_hyperfocuser(self.profiler, focuser),
+                },
+                websocket=ws,
+            )
+
+    async def send_hyperfocus_del(self, profileid: int, websocket=None):
+        await self.send_event(
+            {
+                "type": "DO",
+                "method": "delFocus",
+                "args": {"userid": profileid},
+            },
+            websocket=websocket,
+        )
+
+    async def send_hyperfocus_put(
+        self, communityid: int, focusers: list[Hyperfocuser], websocket=None
+    ):
+        payload = []
+        for focuser in focusers:
+            fpl = await prepare_hyperfocuser(self.profiler, focuser)
+            payload.append(fpl)
+
+        for ws in (websocket,) if websocket else self.communities[communityid]:
+            await self.send_event(
+                {"type": "DO", "method": "putFocus", "args": payload},
+                websocket=ws,
+            )
+
+    async def send_event(self, event, **kwargs):
+        logger.info(f"Sending websocket event: {json.dumps(event, indent=1)}")
+        await super().send_event(event, **kwargs)