diff --git a/k8s/tasks.yml b/k8s/tasks.yml index 0b934cc..8ac698d 100644 --- a/k8s/tasks.yml +++ b/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" diff --git a/payphone/cogs/calls/cog.py b/payphone/cogs/calls/cog.py index 7040df6..70cc67e 100644 --- a/payphone/cogs/calls/cog.py +++ b/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, diff --git a/payphone/core/constants.py b/payphone/core/constants.py index 5295c69..591cda2 100644 --- a/payphone/core/constants.py +++ b/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.""" diff --git a/procs/flush_messages.py b/procs/flush_messages.py index 26b8a7e..fdf9c14 100644 --- a/procs/flush_messages.py +++ b/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() diff --git a/tasks.py b/tasks.py index 404b4f0..afb5041 100644 --- a/tasks.py +++ b/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()