feat(bot): ✨ Migrate tasks to separate containers and readd support for automatic call hangup and account deletion.

k8s/tasks.yml

@@ -19,7 +19,7 @@ spec:
           resources:
             requests:
               memory: "250Mib"
-              cpu: "1"
+              cpu: "100m"
             limits:
               memory: "1Gi"
               cpu: "1"
@@ -29,7 +29,17 @@ spec:
           resources:
             requests:
               memory: "250Mib"
+              cpu: "100m"
+            limits:
+              memory: "1Gi"
               cpu: "1"
+        - name: hangup-inactive-calls
+          image: git.nyeki.dev/payphone/payphone-tasks:latest
+          command: ["uv", "run", "--no-dev", "tasks.py", "hangup-inactive-calls"]
+          resources:
+            requests:
+              memory: "250Mib"
+              cpu: "100m"
             limits:
               memory: "1Gi"
               cpu: "1"

payphone/cogs/calls/cog.py

@@ -24,62 +24,9 @@ class Calls(base.Cog):
     def __init__(self, bot_instance: bot.Payphone):
         self.bot = bot_instance
 
-        self.hangup_calls.add_exception_type(exceptions.ConfigurationError)
-        # self.hangup_calls.start()
-
     def cog_unload(self) -> None:
-        self.hangup_calls.cancel()
-
         return super().cog_unload()
 
-    @tasks.loop(seconds=15)
-    async def hangup_calls(self) -> None:
-        """Hangs up inactive calls.
-
-        Hangs up inactive calls. Calls are hung up when they either:
-        - Have had no messages and more than one minute has passed since the start of the call.
-        - Have had no messages for the past 15 minutes.
-
-        In case they are hung up, both channels are attempted to be notified. In case of failure,
-        that channel is not notified and hung up silently.
-        """
-
-        calls = await models.Call.filter(ended_at=None).prefetch_related("channels")
-
-        hung_up_channel_ids = []
-
-        for call in calls:
-            message = (
-                await models.Message.filter(call=call).order_by("-id").limit(1).first()
-            )
-
-            if (
-                message is None
-                and call.started_at < datetime.now(timezone.utc) - timedelta(minutes=1)
-            ) or (
-                message is not None
-                and message.created_at
-                < datetime.now(timezone.utc) - timedelta(minutes=15)
-            ):
-                for channel in call.channels:
-                    hung_up_channel_ids.append(channel.id)
-
-                    try:
-                        await self.bot.get_partial_messageable(channel.id).send(
-                            "☎️ The call was hung up because nobody replied."
-                        )
-
-                    except errors.DiscordException:
-                        pass
-
-        hung_up_channel_ids = [id for id in hung_up_channel_ids if id is not None]
-
-        if len(hung_up_channel_ids) > 0:
-            await models.Channel.filter(id__in=hung_up_channel_ids).update(call_id=None)
-            await self.bot.redis.srem(
-                constants.CHANNELS_ONCALL_SET, *hung_up_channel_ids
-            )
-
     @commands.command(aliases=["c"])
     @commands.guild_only()
     @commands.cooldown(1, 15, commands.BucketType.channel)
@@ -811,6 +758,8 @@ class Calls(base.Cog):
 
         pipe = self.bot.redis.pipeline()
 
+        pipe.set(constants.CALL_IS_ACTIVE % call.id, "", ex=60 * 15)
+
         pipe.hset(
             constants.CALL_MESSAGES_TEMPLATE % call.id,
             message.id,

payphone/core/constants.py

@@ -211,6 +211,12 @@ CALL_MESSAGE_REPLIES_TEMPLATE = "mr:%s"
 A map of message IDs to a `,`-delimited list of message IDs that reply to the message.
 """
 
+CALL_IS_ACTIVE = "cis:%s"
+"""`%s`: Call ID.
+
+An expiring key that indicates whether the call is still active or not. This key will exist while
+the call is still active."""
+
 CALL_IDS_TO_FLUSH = "call-ids-to-flush"
 """A set of IDs of calls to flush to the DB."""
 

procs/flush_messages.py

@@ -65,6 +65,7 @@ async def flush():
             constants.CALL_MESSAGE_FROM_PROPAGATION_TEMPLATE % call_id,
             constants.CALL_MESSAGES_TEMPLATE % call_id,
             constants.CALL_MESSAGE_REPLIES_TEMPLATE % call_id,
+            constants.CALL_IS_ACTIVE % call_id
         )
 
         await pipe.execute()

tasks.py

@@ -1,3 +1,5 @@
+from datetime import datetime, timezone, timedelta
+
 import asyncio
 
 import functools
@@ -7,11 +9,14 @@ import click
 from procs import flush_messages
 
 from payphone.db import setup, models
-from payphone.core import logging
+from payphone.bot import bot
+from payphone.core import logging, config, constants
 
 
 logging.setup("Tasks")
 
+log = logging.logger
+
 
 def coroutine(f):
     @functools.wraps(f)
@@ -43,9 +48,60 @@ async def delete_accounts():
     """Delete all accounts that were scheduled for deletion. Runs every 1 hour."""
 
     while True:
-        await models.User.filter(is_deleting=True).delete()
+        total = await models.User.filter(is_deleting=True).delete()
+
+        log.info("Deleted %s accounts scheduled for deletion.", total)
+
         await asyncio.sleep(60 * 60)
 
 
+@main.command()
+@coroutine
+async def hangup_inactive_calls():
+    """Hangs up inactive calls."""
+
+    await bot.login(config.TOKEN)
+
+    await bot.redis.ping()
+
+    while True:
+        await asyncio.sleep(5)
+
+        for call in await models.Call.filter(ended_at=None).prefetch_related(
+            "channels"
+        ):
+            if call.started_at > datetime.now(timezone.utc) - timedelta(minutes=1):
+                continue
+
+            if (
+                await bot.redis.exists(
+                    constants.CALL_MESSAGES_TEMPLATE % call.id,
+                    constants.CALL_IS_ACTIVE % call.id,
+                )
+            ) != 2:
+                call.ended_at = datetime.now(timezone.utc)
+                await call.save()
+
+                log.info("Hung up call %s.", call.id)
+
+                try:
+                    channels = [
+                        await bot.get_or_fetch_channel(channel.id)
+                        for channel in call.channels
+                    ]
+
+                    for channel in channels:
+                        try:
+                            await channel.send(
+                                "☎️ The call was hung up because nobody replied."
+                            )
+
+                        finally:
+                            pass
+
+                finally:
+                    continue
+
+
 if __name__ == "__main__":
     main()