feat: add more debugging routes

suggestions/bot.py

@@ -2,6 +2,7 @@
 
 import asyncio
 import datetime
+import io
 import logging
 import math
 import os
@@ -571,6 +572,7 @@ async def load(self):
         await self.stats.load()
         await self.update_bot_listings()
         await self.push_status()
+        await self.update_dev_channel()
         await self.watch_for_shutdown_request()
         await self.load_cogs()
         await self.zonis.start()
@@ -635,6 +637,61 @@ async def process_watch_for_shutdown():
         process_watch_for_shutdown.__task = task_1
         state.add_background_task(task_1)
 
+    async def update_dev_channel(self):
+        if not self.is_prod:
+            log.info("Not watching for debug info as not on prod")
+            return
+
+        state: State = self.state
+
+        async def process_watch_for_shutdown():
+            await self.wait_until_ready()
+            log.debug("Started tracking bot latency")
+
+            while not state.is_closing:
+                # Update once an hour
+                await self.sleep_with_condition(
+                    datetime.timedelta(minutes=5).total_seconds(),
+                    lambda: self.state.is_closing,
+                )
+
+                await self.garven.notify_devs(
+                    title=f"WS latency as follows",
+                    description=f"Timestamped for {datetime.datetime.utcnow().isoformat()}",
+                    sender=f"N/A",
+                )
+
+                data = await self.garven.get_bot_ws_latency()
+                shard_data = data["shards"]
+                for i in range(0, 75, 5):
+                    description = io.StringIO()
+                    for o in range(0, 6):
+                        shard = str(i + o)
+                        try:
+                            description.write(
+                                f"**Shard {shard}**\nWS latency: `{shard_data[shard]['ws']}`\n"
+                                f"Keep Alive latency: `{shard_data[shard]['keepalive']}`\n\n"
+                            )
+                        except KeyError:
+                            # My lazy way of not doing env checks n math right
+                            continue
+
+                    if description.getvalue():
+                        await self.garven.notify_devs(
+                            title=f"WS latency",
+                            description=description.getvalue(),
+                            sender=f"Partial response: {data['partial_response']}",
+                        )
+
+                await self.sleep_with_condition(
+                    datetime.timedelta(hours=1).total_seconds(),
+                    lambda: self.state.is_closing,
+                )
+
+        task_1 = asyncio.create_task(process_watch_for_shutdown())
+        process_watch_for_shutdown.__task = task_1
+        state.add_background_task(task_1)
+
     async def update_bot_listings(self) -> None:
         """Updates the bot lists with current stats."""
         if not self.is_prod:

suggestions/garven.py

@@ -21,11 +21,13 @@ def __init__(self, bot: SuggestionsBot):
             if bot.is_prod
             else "https://garven.dev.suggestions.gg"
         )
+        # self._url = "http://127.0.0.1:8002"
         self._ws_url = (
             "wss://garven.suggestions.gg/ws"
             if bot.is_prod
             else "wss://garven.dev.suggestions.gg/ws"
         )
+        # self._ws_url = "ws://127.0.0.1:8002/ws"
         self._session: aiohttp.ClientSession = aiohttp.ClientSession(
             base_url=self._url,
             headers={"X-API-KEY": os.environ["GARVEN_API_KEY"]},
@@ -85,3 +87,10 @@ async def cluster_status(self) -> dict:
             data = await resp.json()
 
         return data
+
+    async def get_bot_ws_latency(self):
+        async with self._session.get("/cluster/latency/ws") as resp:
+            await self._handle_status(resp)
+            data = await resp.json()
+
+        return data

suggestions/zonis_routes.py

@@ -3,7 +3,7 @@
 import logging
 import math
 import os
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Literal
 
 import disnake
 from zonis import client
@@ -33,6 +33,7 @@ def __init__(self, bot: SuggestionsBot):
             "refresh_premium",
             "shared_guilds",
             "cached_item_count",
+            "cluster_ws_status",
         )
 
     async def start(self):
@@ -65,6 +66,25 @@ async def cluster_status(self):
 
         return data
 
+    @client.route()
+    async def cluster_ws_status(
+        self,
+    ) -> dict[str, dict[Literal["ws", "keepalive"], str]]:
+        data: dict[str, dict[Literal["ws", "keepalive"], str]] = {}
+        for shard_id, shard_info in self.bot.shards.items():
+            shard_data: dict[Literal["ws", "keepalive"], str] = {
+                "ws": str(round(shard_info.latency, 5))
+            }
+            wsc = shard_info._parent.ws._keep_alive
+            if wsc is None:
+                shard_data["keepalive"] = "None"
+            else:
+                shard_data["keepalive"] = str(round(wsc.latency, 5))
+
+            data[str(shard_id)] = shard_data
+
+        return data
+
     @client.route()
     async def share_with_devs(self, title, description, sender):
         channel: disnake.TextChannel = await self.bot.get_or_fetch_channel(  # type: ignore