diff --git a/Cargo.lock b/Cargo.lock index 4f045ec..1604f19 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1990,7 +1990,7 @@ checksum = "df94ce210e5bc13cb6651479fa48d14f601d9858cfe0467f43ae157023b938d3" [[package]] name = "payphone" -version = "4.3.7" +version = "4.3.10" dependencies = [ "anyhow", "async-trait", diff --git a/Cargo.toml b/Cargo.toml index f336dda..5b63abe 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "payphone" -version = "4.3.7" +version = "4.3.10" edition = "2024" default-run = "payphone" diff --git a/src/framework/bot.rs b/src/framework/bot.rs index f0cf459..326745f 100644 --- a/src/framework/bot.rs +++ b/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 { +pub struct Bot { shard: ShardId, - state: Arc, + state: Arc, presence: Option, - prefixes: Arc + Send + Sync>, + prefixes: Arc + Send + Sync>, intents: Intents, - extras: Arc>, + extras: Arc>, cached_resource_types: Vec, - commands: Vec>>, - components: Vec>>, - on_ready_handlers: Vec>>, - on_reconnect_handlers: Vec>>, - on_interaction_handlers: Vec>>, - on_message_handlers: Vec>>, - on_message_edit_handlers: Vec>>, - on_message_delete_handlers: Vec>>, - on_server_create_handlers: Vec>>, - on_server_update_handlers: Vec>>, - on_server_delete_handlers: Vec>>, + commands: Vec>>, + components: Vec>>, + tasks: Vec>>, + on_ready_handlers: Vec>>, + on_reconnect_handlers: Vec>>, + on_interaction_handlers: Vec>>, + on_message_handlers: Vec>>, + on_message_edit_handlers: Vec>>, + on_message_delete_handlers: Vec>>, + on_server_create_handlers: Vec>>, + on_server_update_handlers: Vec>>, + on_server_delete_handlers: Vec>>, } impl Bot @@ -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 + '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 { diff --git a/src/framework/mod.rs b/src/framework/mod.rs index 1b19238..6d35f24 100644 --- a/src/framework/mod.rs +++ b/src/framework/mod.rs @@ -161,3 +161,4 @@ pub mod events; pub mod interactions; pub mod messages; pub mod runtime; +pub mod tasks; diff --git a/src/framework/tasks.rs b/src/framework/tasks.rs new file mode 100644 index 0000000..3794bfe --- /dev/null +++ b/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: Send + Sync + Debug { + fn delay(&self) -> Duration; + + async fn run(&self, rt: &Runtime) -> anyhow::Result<()>; +} diff --git a/src/groups/calls/commands.rs b/src/groups/calls/commands.rs index 4374da5..a6a18fb 100644 --- a/src/groups/calls/commands.rs +++ b/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 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 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 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 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 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:?}"))?; diff --git a/src/groups/calls/events.rs b/src/groups/calls/events.rs index 5f0779d..f9104ce 100644 --- a/src/groups/calls/events.rs +++ b/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 for CallMessageHandler { .icon_url(ImageSource::url(avatar_url).unwrap()) .url(message_link), ) - .color( - if replying_to.data.author_mask.is_some() { - COLOR_WHITE - } else { - replying_to_user + .color(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 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) diff --git a/src/groups/calls/mod.rs b/src/groups/calls/mod.rs index 425d88b..261d319 100644 --- a/src/groups/calls/mod.rs +++ b/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; diff --git a/src/groups/calls/tasks.rs b/src/groups/calls/tasks.rs new file mode 100644 index 0000000..053300b --- /dev/null +++ b/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 for EndInactiveCallsTask { + fn delay(&self) -> std::time::Duration { + std::time::Duration::from_secs(5) + } + + async fn run(&self, rt: &Runtime) -> 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::().ok(); + let b = b.parse::().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(()) + } +} diff --git a/src/main.rs b/src/main.rs index 257b5f2..ee097be 100644 --- a/src/main.rs +++ b/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)