Rewrite a v2 of the Mentor Requests cog

IsaacG · IsaacG · commit a49691f903a8 · 2025-10-29T01:08:33.000-07:00
diff --git a/cogs/mentor_requests_v2.md b/cogs/mentor_requests_v2.md
@@ -0,0 +1,38 @@
+# Mentor Requests Cog
+
+## Overview
+
+This cog polls Exercism's API for requests in the queue.
+The list of requests is synced to a per-track thread in the mentor requests Discord channel.
+
+## Periodic Tasks
+
+* Poll Exercism for mentor requests.
+  * Each track can be handled as a separate task vs polling all tracks every time we get an updated.
+    This is helpful since some tracks are significantly more active than others.
+    We want to poll at most once every 5 minutes.
+    Less active tracks can be polled as little as once per hour.
+  * Store a copy of this in the DB and in memory.
+  * Expose Exercism request rates to Prometheus.
+  * Maintain the timestamps of the past N requests per-track to set the per-track interval.
+    Spitballing, maybe use avg - 1 * stddev, clamped to 5-60 minutes.
+* Poll Discord to get all messages in the channel/threads.
+  * Since we control the messages, they shouldn't drift out of sync too often.
+  * Reading messages from Discord should be relatively light weight.
+  * Spread the reads. One track per minute.
+  * Store the results in the DB and in memory.
+* On any state change, queue the change (add message, remove message).
+
+## Tasks
+
+* Fetch track requests from Exercism.
+* Fetch Discord messages for a track.
+* Send a Discord message.
+* Delete a Discord message.
+
+## Worker
+
+* Use a loop task that runs every 5 seconds.
+* Use an async-safe lock so only one task runs at a time. If the lock is held, return.
+* Store the timestamp for the next queued task. If the timestamp is in the future, return.
+* If there is any issues executing a task, leave it for the next loop.
diff --git a/cogs/mentor_requests_v2.py b/cogs/mentor_requests_v2.py
@@ -0,0 +1,310 @@
+"""Discord module to publish mentor request queues. Version 2."""
+
+import asyncio
+import datetime
+import enum
+import logging
+import random
+import re
+import statistics
+import time
+from typing import Sequence
+
+import discord
+import prometheus_client  # type: ignore
+from discord.ext import commands
+from discord.ext import tasks
+from exercism_lib import exercism
+
+from cogs import base_cog
+
+logger = logging.getLogger(__name__)
+
+PROM_EXERCISM_REQUESTS = prometheus_client.Counter(
+    "mentor_request_exercism_rpc", "Number of API calls to Exercism", ["track"]
+)
+PROM_DISCORD_REQUESTS = prometheus_client.Counter(
+    "mentor_request_discord_rpc", "Number of API calls to Discord", ["write"]
+)
+PROM_TASK_QUEUE = prometheus_client.Gauge("mentor_requests_task_queue", "size of the task queue")
+
+
+class TaskType(enum.IntEnum):
+    """Types of tasks to execute."""
+    TASK_QUERY_EXERCISM = enum.auto()
+    TASK_QUERY_DISCORD = enum.auto()
+    TASK_DISCORD_ADD = enum.auto()
+    TASK_DISCORD_DEL = enum.auto()
+
+
+class RequestNotifierV2(base_cog.BaseCog):
+    """Update Discord with Mentor Requests."""
+
+    qualified_name = "Request Notifier v2"
+
+    def __init__(
+        self,
+        bot: commands.Bot,
+        channel_id: int,
+        tracks: Sequence[str] | None = None,
+        **kwargs,
+    ) -> None:
+        super().__init__(bot=bot, **kwargs)
+        self.exercism = exercism.AsyncExercism()
+        self.channel_id = channel_id
+
+        self.threads: dict[str, discord.Thread] = {}
+        self.requests: dict[str, tuple[str, discord.Message]] = {}
+        self.messages: dict[str, dict[str, str]] = {}
+        self.next_task_time = 0
+
+        self.lock = asyncio.Lock()
+        self.queue = asyncio.PriorityQueue()
+
+        if tracks:
+            self.tracks = list(tracks)
+        else:
+            self.tracks = exercism.Exercism().all_tracks()
+        self.tracks.sort()
+        # Default to 10 minute polling.
+        self.request_interval = {track: 600 for track in self.tracks}
+        self.request_timestamps = {track: [] for track in self.tracks}
+
+        self.task_manager.start()  # pylint: disable=E1101
+
+    async def get_thread(self, track: str) -> discord.Thread:
+        """Return the request thread for a specific track."""
+        thread = self.threads.get(track)
+        if not thread:
+            raise LookupError(f"Failed to find track {track} in threads")
+
+        # Refresh the thread object. This is helpful to update the is_archived bit.
+        async with asyncio.timeout(10):
+            got = await self.bot.fetch_channel(thread.id)
+        assert isinstance(got, discord.Thread), f"Expected a Thread. {got=}"
+        self.threads[track] = got
+        return got
+
+    @tasks.loop(seconds=5)
+    async def task_manager(self):
+        """Task loop."""
+        if self.lock.locked():
+            return
+        async with self.lock:
+            try:
+                PROM_TASK_QUEUE.set(self.queue.qsize())
+                now = int(time.time())
+                # If the queue is empty or the next task is not yet due, return.
+                if now < self.next_task_time or self.queue.empty():
+                    return
+                task = self.queue.get_nowait()
+                task_time, task_type, track, *details = task
+                # If the next task is not yet due, queue it and return.
+                if now < task_time:
+                    self.queue.put_nowait(task)
+                    self.next_task_time = task_time
+                    return
+                # Handle a task.
+                if task_type == TaskType.TASK_QUERY_EXERCISM:
+                    try:
+                        await self.fetch_track_requests(track)
+                    finally:
+                        self.queue_query_exercism(track)
+                elif task_type == TaskType.TASK_QUERY_DISCORD:
+                    try:
+                        await self.fetch_discord_thread(track)
+                    finally:
+                        self.queue_query_discord(track)
+                elif task_type == TaskType.TASK_DISCORD_ADD:
+                    await self.update_discord_add(track, *details)
+                elif task_type == TaskType.TASK_DISCORD_DEL:
+                    await self.update_discord_del(track, *details)
+                else:
+                    logger.exception("Unknown task type, %d", task_type)
+
+            except Exception:  # pylint: disable=broad-exception-caught
+                logger.exception("Unhandled exception in task manager loop.")
+
+    @task_manager.before_loop
+    async def before_task_manager(self):
+        """Before starting the task manager, wait for ready and load Discord messages."""
+        logger.debug("Start before_task_manager()")
+        await self.bot.wait_until_ready()
+        await self.load_data()
+        self.populate_task_queue()
+        logger.debug("End before_task_manager()")
+
+    def exercism_poll_interval(self, track: str) -> int:
+        """Return the poll interval between getting requests for a track."""
+        interval = self.request_interval[track]
+        times = self.request_timestamps[track]
+        if len(times) < 2:
+            self.request_interval[track] = min(int(1.5 * interval), 60 * 60 * 60)
+            return interval
+        times.sort()
+        intervals = [a - b for a, b in zip(times[1:], times)]
+        # [5 min ... avg ... 1 hour]
+        return min(max(int(statistics.mean(intervals)), 60 * 5), 60 * 60 * 60)
+
+    async def fetch_track_requests(self, track: str) -> None:
+        """Fetch the requests for a given track. Queue tasks to update the Discord thread."""
+        logger.debug("Start fetch_track_requests(%s)", track)
+        if track not in self.messages:
+            return
+        PROM_EXERCISM_REQUESTS.labels(track).inc()
+        # fetch requests
+        # update DB
+        # update request interval data
+        # compare to Discord thread; queue tasks to add/remove.
+        async with asyncio.timeout(15):
+            requests = await self.get_requests(track)
+        logger.debug("Found %d requests for %s.", len(requests), track)
+
+        add_requests = set(requests) - set(self.messages[track])
+        del_requests = set(self.messages[track]) - set(requests)
+        self.requests[track] = {
+            request_id: message
+            for request_id, (timestamp, message) in requests.items()
+        }
+        self.request_timestamps[track].extend(
+            timestamp
+            for request_id, (timestamp, message) in requests.items()
+            if request_id in add_requests
+        )
+        self.request_timestamps[track] = sorted(self.request_timestamps[track], reverse=True)[:10]
+
+        for request_id in add_requests:
+            logger.debug("Queue TASK_DISCORD_ADD %s %s for now", track, request_id)
+            self.queue.put_nowait((0, TaskType.TASK_DISCORD_ADD, track, request_id))
+
+        if del_requests:
+            to_del = tuple(list(del_requests)[:10])
+            logger.debug("Queue TASK_DISCORD_DEL %s %s for now", track, to_del)
+            self.queue.put_nowait((0, TaskType.TASK_DISCORD_DEL, track, to_del))
+
+    def queue_query_exercism(self, track) -> None:
+        """Queue a task to query Exercism for a track."""
+        interval = self.exercism_poll_interval(track)
+        task_time = int(time.time()) + interval
+        logger.debug("Queue TASK_QUERY_EXERCISM %s in %d seconds", track, interval)
+        self.queue.put_nowait((task_time, TaskType.TASK_QUERY_EXERCISM, track))
+
+    async def fetch_discord_thread(self, track) -> None:
+        """Fetch a track thread from Discord to update the local cache."""
+        logger.debug("Start fetch_discord_thread(%s)", track)
+        request_url_re = re.compile(r"\bhttps://exercism.org/mentoring/requests/(\w+)\b")
+        PROM_DISCORD_REQUESTS.labels(False).inc()
+        thread = await self.get_thread(track)
+        messages = {}
+        await self.unarchive(thread)
+        async for message in thread.history():
+            if message.author != thread.owner:
+                continue
+            if message == thread.starter_message:
+                continue
+            match = request_url_re.search(message.content)
+            if match is None:
+                continue
+            messages[match.group(1)] = message.id
+        self.messages[track] = messages
+
+    def queue_query_discord(self, track: str) -> None:
+        """Queue a task to query a Discord request thread."""
+        interval = 60  # one minute
+        task_time = int(time.time()) + interval
+        logger.debug("Queue TASK_QUERY_DISCORD %s in %d seconds", track, interval)
+        self.queue.put_nowait((task_time, TaskType.TASK_QUERY_DISCORD, track))
+
+    async def update_discord_add(self, track: str, request_id: str) -> None:
+        """Add a request message to Discord."""
+        logger.debug("Start update_discord_add(%s, %s)", track, request_id)
+        PROM_DISCORD_REQUESTS.labels(True).inc()
+        thread = await self.get_thread(track)
+        description = self.requests[track][request_id]
+        async with asyncio.timeout(10):
+            message = await thread.send(description, suppress_embeds=True)
+        self.messages[track][request_id] = message.id
+
+    async def update_discord_del(self, track: str, message_ids: tuple[int, ...]) -> None:
+        """Remove a request message from Discord."""
+        PROM_DISCORD_REQUESTS.labels(True).inc()
+        logger.debug("Start update_discord_del(%s, %s)", track, message_ids)
+        thread = await self.get_thread(track)
+        await self.unarchive(self.threads[track])
+
+        async with asyncio.timeout(15):
+            try:
+                await thread.delete_messages(
+                    discord.Object(message_id) for message_id in message_ids
+                )
+            except discord.errors.NotFound:
+                pass
+        request_ids = [
+            request_id
+            for request_id, message_id in self.messages[track].items()
+            if message_id in message_ids
+        ]
+        for request_id in request_ids:
+            del self.messages[track][request_id]
+
+    def populate_task_queue(self):
+        """Populate the initial task queue."""
+        tracks = self.tracks.copy()
+        random.shuffle(tracks)
+        # Spread the initial requests over 5 minutes
+        for track, offset in zip(tracks, range(0, 5 * 60, int(5 * 60 / len(tracks)))):
+            task_time = int(time.time()) + offset
+            self.queue.put_nowait((task_time, TaskType.TASK_QUERY_DISCORD, track))
+            self.queue.put_nowait((task_time + 1, TaskType.TASK_QUERY_EXERCISM, track))
+
+    async def unarchive(self, thread: discord.Thread) -> None:
+        """Ensure a thread is not archived."""
+        if not thread.archived:
+            return
+        async with asyncio.timeout(10):
+            message = await thread.send("Sending a message to unarchive this thread.")
+        async with asyncio.timeout(10):
+            await message.delete()
+
+    async def load_data(self) -> None:
+        """Load Exercism data."""
+        channel = self.bot.get_channel(self.channel_id)
+        assert isinstance(channel, discord.TextChannel), f"{channel} is not a TextChannel."
+
+        self.threads = {}
+        async for message in channel.history():
+            if not message.thread:
+                continue
+            thread = await message.fetch_thread()
+            self.threads[thread.name.lower()] = thread
+
+        for track in self.tracks:
+            if track in self.threads:
+                continue
+            thread = await channel.create_thread(
+                name=track.title(),
+                type=discord.ChannelType.public_thread,
+            )
+            self.threads[track] = thread
+            await asyncio.sleep(2)
+
+    async def get_requests(self, track_slug: str) -> dict[str, tuple[int, str]]:
+        """Return formatted mentor requests."""
+        requests = {}
+        for req in await self.exercism.mentor_requests(track_slug):
+            # uuid = req["uuid"]
+            track_title = req["track"]["title"]
+            exercise_title = req["exercise"]["title"]
+            student_handle = req["student"]["handle"]
+            status = req["status"]
+            url = req["url"]
+
+            msg = f"{track_title.title()}: {url} => {exercise_title} "
+            if status:
+                msg += f"({student_handle}, {status})"
+            else:
+                msg += f"({student_handle})"
+
+            timestamp = int(datetime.datetime.fromisoformat(req["updated_at"]).timestamp())
+            requests[req["uuid"]] = (timestamp, msg)
+        return requests
