Expire calls where one of both parties doesn't talk for 10 minutes.

Cargo.lock

@@ -1990,7 +1990,7 @@ checksum = "df94ce210e5bc13cb6651479fa48d14f601d9858cfe0467f43ae157023b938d3"
 
 [[package]]
 name = "payphone"
-version = "4.3.7"
+version = "4.3.10"
 dependencies = [
  "anyhow",
  "async-trait",

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "payphone"
-version = "4.3.7"
+version = "4.3.10"
 edition = "2024"
 default-run = "payphone"
 

src/framework/bot.rs

@@ -19,6 +19,7 @@ use crate::{
             components::Component, registry::ComponentRegistry, runner::InteractionRunner,
         },
         runtime::Runtime,
+        tasks::Task,
     },
     utils::Bye,
 };
@@ -92,25 +93,26 @@ where
     }
 }
 
-pub struct Bot<T: Send + Sync, E: Send + Sync = ()> {
+pub struct Bot<S: Send + Sync, E: Send + Sync = ()> {
     shard: ShardId,
-    state: Arc<T>,
+    state: Arc<S>,
     presence: Option<UpdatePresencePayload>,
-    prefixes: Arc<dyn PrefixGetter<T, E> + Send + Sync>,
+    prefixes: Arc<dyn PrefixGetter<S, E> + Send + Sync>,
     intents: Intents,
-    extras: Arc<dyn ExtrasGetter<T, Extras = E>>,
+    extras: Arc<dyn ExtrasGetter<S, Extras = E>>,
     cached_resource_types: Vec<ResourceType>,
-    commands: Vec<Arc<dyn Command<T, E>>>,
+    commands: Vec<Arc<dyn Command<S, E>>>,
-    components: Vec<Arc<dyn Component<T, E>>>,
+    components: Vec<Arc<dyn Component<S, E>>>,
-    on_ready_handlers: Vec<Arc<dyn OnReadyHandler<T, E>>>,
+    tasks: Vec<Arc<dyn Task<S, E>>>,
-    on_reconnect_handlers: Vec<Arc<dyn OnReconnectHandler<T, E>>>,
+    on_ready_handlers: Vec<Arc<dyn OnReadyHandler<S, E>>>,
-    on_interaction_handlers: Vec<Arc<dyn OnInteractionHandler<T, E>>>,
+    on_reconnect_handlers: Vec<Arc<dyn OnReconnectHandler<S, E>>>,
-    on_message_handlers: Vec<Arc<dyn OnMessageHandler<T, E>>>,
+    on_interaction_handlers: Vec<Arc<dyn OnInteractionHandler<S, E>>>,
-    on_message_edit_handlers: Vec<Arc<dyn OnMessageEditHandler<T, E>>>,
+    on_message_handlers: Vec<Arc<dyn OnMessageHandler<S, E>>>,
-    on_message_delete_handlers: Vec<Arc<dyn OnMessageDeleteHandler<T, E>>>,
+    on_message_edit_handlers: Vec<Arc<dyn OnMessageEditHandler<S, E>>>,
-    on_server_create_handlers: Vec<Arc<dyn OnServerCreateHandler<T, E>>>,
+    on_message_delete_handlers: Vec<Arc<dyn OnMessageDeleteHandler<S, E>>>,
-    on_server_update_handlers: Vec<Arc<dyn OnServerUpdateHandler<T, E>>>,
+    on_server_create_handlers: Vec<Arc<dyn OnServerCreateHandler<S, E>>>,
-    on_server_delete_handlers: Vec<Arc<dyn OnServerDeleteHandler<T, E>>>,
+    on_server_update_handlers: Vec<Arc<dyn OnServerUpdateHandler<S, E>>>,
+    on_server_delete_handlers: Vec<Arc<dyn OnServerDeleteHandler<S, E>>>,
 }
 
 impl<T, E> Bot<T, E>
@@ -133,6 +135,7 @@ where
             cached_resource_types: Vec::new(),
             commands: Vec::new(),
             components: Vec::new(),
+            tasks: Vec::new(),
             on_ready_handlers: Vec::new(),
             on_reconnect_handlers: Vec::new(),
             on_interaction_handlers: vec![Arc::new(InteractionRunner)],
@@ -170,6 +173,11 @@ where
         self
     }
 
+    pub fn task(mut self, task: impl Task<T, E> + 'static) -> Self {
+        self.tasks.push(Arc::new(task));
+        self
+    }
+
     pub fn cache_resource_type(mut self, resource_type: ResourceType) -> Self {
         if !self.cached_resource_types.contains(&resource_type) {
             self.cached_resource_types.push(resource_type);
@@ -273,6 +281,23 @@ where
             cache,
         });
 
+        for task in bot.tasks.clone() {
+            let task_rt = runtime.clone();
+
+            tokio::spawn(async move {
+                loop {
+                    task.run(&task_rt)
+                        .await
+                        .inspect_err(|e| {
+                            log::error!("An error occurred while running task {task:?}! {e:?}")
+                        })
+                        .bye();
+
+                    tokio::time::sleep(task.delay()).await;
+                }
+            });
+        }
+
         let mut config = ConfigBuilder::new(token.to_string(), bot.intents);
 
         if let Some(presence) = &bot.presence {

src/framework/mod.rs

@@ -161,3 +161,4 @@ pub mod events;
 pub mod interactions;
 pub mod messages;
 pub mod runtime;
+pub mod tasks;

src/framework/tasks.rs

@@ -0,0 +1,12 @@
+use std::{fmt::Debug, time::Duration};
+
+use async_trait::async_trait;
+
+use crate::framework::runtime::Runtime;
+
+#[async_trait]
+pub trait Task<S, E>: Send + Sync + Debug {
+    fn delay(&self) -> Duration;
+
+    async fn run(&self, rt: &Runtime<S, E>) -> anyhow::Result<()>;
+}

src/groups/calls/commands.rs

@@ -1,5 +1,7 @@
 use async_trait::async_trait;
+use deadpool_redis::redis::AsyncTypedCommands;
 use scc::HashSet;
+use time::UtcDateTime;
 use twilight_model::id::Id;
 
 use crate::{
@@ -115,12 +117,32 @@ impl Command<State, Extras> for CallCommand {
                 ))
                 .await
                 .bye();
+
             ctx.rt
                 .http
                 .create_message(Id::new(call.other_party_channel_id(channel.id)))
                 .content(&format!("☎️ {}", get_string_call_answered(ctx.prefix)))
                 .await
                 .bye();
+
+            let mut redis = ctx.rt.state.redis.get().await?;
+
+            let timestamp = UtcDateTime::now().unix_timestamp();
+
+            redis
+                .zadd(
+                    "active-calls",
+                    format!("{}:{}", call.id, channel.id),
+                    timestamp,
+                )
+                .await?;
+            redis
+                .zadd(
+                    "active-calls",
+                    format!("{}:{}", call.id, call.other_party_channel_id(channel.id)),
+                    timestamp,
+                )
+                .await?;
         } else {
             ctx.send()
                 .content(format!(
@@ -215,6 +237,19 @@ impl Command<State, Extras> for SkipCommand {
                 log::error!("Failed to send message to other party: {e:?}");
             })?;
 
+        let mut redis = ctx.rt.state.redis.get().await?;
+
+        let channel_keys = [
+            format!("{}:{}", call.id, call.channel_ids[0]),
+            format!("{}:{}", call.id, call.channel_ids[1]),
+        ];
+
+        redis
+            .zrem("active-calls", &channel_keys)
+            .await
+            .inspect_err(|e| log::error!("Failed to remove inactive call from redis: {e}"))
+            .ok();
+
         let call = Call::start(&channel, &ctx.rt.state.db)
             .await
             .inspect_err(|e| {
@@ -236,6 +271,25 @@ impl Command<State, Extras> for SkipCommand {
                 .content(&format!("☎️ {}", get_string_call_answered(ctx.prefix)))
                 .await
                 .bye();
+
+            let mut redis = ctx.rt.state.redis.get().await?;
+
+            let timestamp = UtcDateTime::now().unix_timestamp();
+
+            redis
+                .zadd(
+                    "active-calls",
+                    format!("{}:{}", call.id, channel.id),
+                    timestamp,
+                )
+                .await?;
+            redis
+                .zadd(
+                    "active-calls",
+                    format!("{}:{}", call.id, call.other_party_channel_id(channel.id)),
+                    timestamp,
+                )
+                .await?;
         } else {
             ctx.send()
                 .content(format!(
@@ -304,6 +358,8 @@ impl Command<State, Extras> for HangupCommand {
         }
 
         if let Some(call_id) = channel.call_id {
+            let mut redis = ctx.rt.state.redis.get().await?;
+
             let Some(call) = Call::get_by_id(call_id, &ctx.rt.state.db)
                 .await
                 .inspect_err(|e| log::error!("Failed to get call: {e:?}"))?
@@ -311,11 +367,27 @@ impl Command<State, Extras> for HangupCommand {
                 ctx.send().content("❌ The call could not be found.\n-# This is not supposed to happen. If the issue persists, contact a moderator in the official server.").await?;
 
                 channel.call_id = None;
-                channel.upsert(&ctx.rt.state.db).await?;
+                channel.upsert(&ctx.rt.state.db).await.bye();
+
+                redis
+                    .zrem("active-calls", format!("{}:{}", call_id, channel.id))
+                    .await
+                    .bye();
 
                 return Err(CommandError::DatabaseInconsistentState);
             };
 
+            let channel_keys = [
+                format!("{call_id}:{}", call.channel_ids[0]),
+                format!("{call_id}:{}", call.channel_ids[1]),
+            ];
+
+            redis
+                .zrem("active-calls", &channel_keys)
+                .await
+                .inspect_err(|e| log::error!("Failed to remove inactive call from redis: {e}"))
+                .ok();
+
             call.hangup(&ctx.rt.state.db)
                 .await
                 .inspect_err(|e| log::error!("Failed to hang up call: {e:?}"))?;

src/groups/calls/events.rs

@@ -1,6 +1,7 @@
 use std::{error::Error, sync::Arc};
 
 use async_trait::async_trait;
+use deadpool_redis::redis::AsyncTypedCommands;
 use itertools::Itertools;
 use time::{Duration, UtcDateTime};
 use twilight_http::{api_error::ApiError, error::ErrorType};
@@ -353,15 +354,13 @@ impl OnMessageHandler<State, Extras> for CallMessageHandler {
                                     .icon_url(ImageSource::url(avatar_url).unwrap())
                                     .url(message_link),
                             )
-                            .color(
+                            .color(if replying_to.data.author_mask.is_some() {
-                                if replying_to.data.author_mask.is_some() {
                                 COLOR_WHITE
                             } else {
                                 replying_to_user
                                     .and_then(|i| i.color.map(|c| c as u32))
                                     .unwrap_or(COLOR_WHITE)
-                                }
+                            })
-                            )
                             .description(ellipse(replying_to.data.content, 100))
                             .build(),
                     );
@@ -393,6 +392,16 @@ impl OnMessageHandler<State, Extras> for CallMessageHandler {
                 }
             }
 
+            let mut redis = rt.state.redis.get().await?;
+
+            redis
+                .zadd(
+                    "active-calls",
+                    format!("{}:{}", call.id, channel.id),
+                    UtcDateTime::now().unix_timestamp(),
+                )
+                .await?;
+
             let message_sent = match rt
                 .http
                 .execute_webhook(Id::new(webhook_id), &webhook_token)

src/groups/calls/mod.rs

@@ -8,3 +8,4 @@ pub mod masks;
 pub(super) mod moderation;
 pub(super) mod permissions;
 pub(super) mod strings;
+pub mod tasks;

src/groups/calls/tasks.rs

@@ -0,0 +1,105 @@
+use async_trait::async_trait;
+use deadpool_redis::redis::AsyncTypedCommands;
+use time::{Duration, UtcDateTime};
+use twilight_model::id::Id;
+
+use crate::{
+    database::models::Call,
+    extras::Extras,
+    framework::{runtime::Runtime, tasks::Task},
+    state::State,
+};
+
+#[derive(Debug)]
+pub struct EndInactiveCallsTask {
+    pub current_shard: u32,
+    pub total_shards: u32,
+}
+
+impl EndInactiveCallsTask {
+    pub fn new(current_shard: u32, total_shards: u32) -> Self {
+        Self {
+            current_shard,
+            total_shards,
+        }
+    }
+}
+
+#[async_trait]
+impl Task<State, Extras> for EndInactiveCallsTask {
+    fn delay(&self) -> std::time::Duration {
+        std::time::Duration::from_secs(5)
+    }
+
+    async fn run(&self, rt: &Runtime<State, Extras>) -> anyhow::Result<()> {
+        let mut redis = rt.state.redis.get().await?;
+
+        let calls_and_channels = redis
+            .zrangebyscore(
+                "active-calls",
+                0,
+                (UtcDateTime::now() - Duration::minutes(10)).unix_timestamp(),
+            )
+            .await?;
+
+        let mut ended_call_ids = Vec::new();
+
+        for (call_id, channel_id) in calls_and_channels.iter().filter_map(|i| {
+            i.split_once(":").and_then(|(a, b)| {
+                let a = a.parse::<u64>().ok();
+                let b = b.parse::<u64>().ok();
+
+                a.zip(b)
+            })
+        }) {
+            if ended_call_ids.contains(&call_id) {
+                continue;
+            }
+
+            if (call_id) % self.total_shards as u64 != self.current_shard as u64 {
+                continue;
+            }
+
+            ended_call_ids.push(call_id);
+
+            let Ok(Some(call)) = Call::get_by_id(call_id, &rt.state.db).await else {
+                redis
+                    .zrem("active-calls", format!("{call_id}:{channel_id}"))
+                    .await
+                    .inspect_err(|e| log::error!("Failed to remove inactive call from redis: {e}"))
+                    .ok();
+
+                continue;
+            };
+
+            let channel_keys = [
+                format!("{call_id}:{}", call.channel_ids[0]),
+                format!("{call_id}:{}", call.channel_ids[1]),
+            ];
+
+            redis
+                .zrem("active-calls", &channel_keys)
+                .await
+                .inspect_err(|e| log::error!("Failed to remove inactive call from redis: {e}"))
+                .ok();
+
+            call.hangup(&rt.state.db)
+                .await
+                .inspect_err(|e| log::error!("Failed to hangup inactive call {call_id}: {e}"))
+                .ok();
+
+            for channel_id in call.channel_ids {
+                rt.http
+                    .create_message(Id::new(channel_id))
+                    .content("☎️ The call was hung up because one or both parties were gone for too long.")
+                    .await
+                    .inspect_err(|e| {
+                        log::error!("Failed to send call end message to channel {channel_id}: {e}")
+                    })
+                    .ok();
+            }
+        }
+
+        Ok(())
+    }
+}

src/main.rs

@@ -141,7 +141,7 @@ use crate::{
             commands::{
                 CallCommand, FriendRequestCommand, HangupCommand, MaskCommand, SkipCommand,
             },
-            events::{CallMessageDeleteHandler, CallMessageEditHandler, CallMessageHandler},
+            events::{CallMessageDeleteHandler, CallMessageEditHandler, CallMessageHandler}, tasks::EndInactiveCallsTask,
         },
         gateway::events::{ReadyHandler, ReconnectHandler},
         leveling::{commands::LeaderboardCommand, events::MessageLevelingHandler},
@@ -307,6 +307,7 @@ async fn start(
         .cache_resource_type(ResourceType::CHANNEL)
         .cache_resource_type(ResourceType::MEMBER)
         .cache_resource_type(ResourceType::ROLE)
+        .task(EndInactiveCallsTask::new(shard.number(), shard.total()))
         .on_ready(ReadyHandler)
         .on_reconnect(ReconnectHandler)
         .on_message(CallMessageHandler)