Recreate webhooks when they go missing, reuse older webhooks, and initiate connections to Redis cluster

src/cli.rs

@@ -212,6 +212,15 @@ pub enum Commands {
             default_value_t = Shard::default())
         ]
         shard: Shard,
+
+        /// The Redis URLs to connect to.
+        #[arg(
+            long,
+            env = "PAYPHONE_REDIS_URLS",
+            value_name = "URLS",
+            default_values = &["redis://127.0.0.1:6379"],
+        )]
+        redis_urls: Vec<String>,
     },
 
     /// Starts a documentation server for the project which serves `cargo doc` and refreshes on

src/database/models.rs

@@ -13,7 +13,9 @@ use thiserror::Error;
 use time::{Duration, OffsetDateTime, UtcDateTime, UtcOffset};
 use twilight_http::Client;
 use twilight_model::{
-    channel::Message as TwilightMessage, guild::Guild as TwilightGuild, id::Id,
+    channel::{Message as TwilightMessage, Webhook},
+    guild::Guild as TwilightGuild,
+    id::Id,
     user::User as TwilightUser,
 };
 
@@ -478,6 +480,38 @@ impl<'a> Object<'a> for Channel {
     }
 }
 
+async fn get_or_create_channel_webhook(
+    channel_id: u64,
+    http: &Client,
+) -> Result<Webhook, DatabaseOrDiscordError> {
+    let webhooks = http
+        .channel_webhooks(Id::new(channel_id))
+        .await
+        .inspect_err(|e| log::error!("Failed to get webhooks from channel {channel_id}: {e:?}"))
+        .map_err(DatabaseOrDiscordError::Discord)?
+        .model()
+        .await
+        .inspect_err(|e| log::error!("Failed to get model for webhooks {channel_id}: {e:?}"))
+        .map_err(|_| DatabaseOrDiscordError::DiscordDeserializationFailed)?;
+
+    let webhook = if let Some(item) = webhooks.into_iter().find(|item| item.token.is_some()) {
+        item
+    } else {
+        http.create_webhook(Id::new(channel_id), "Payphone Messages")
+            .await
+            .inspect_err(|e| {
+                log::error!("Failed to create webhook for channel {channel_id}: {e:?}")
+            })
+            .map_err(DatabaseOrDiscordError::Discord)?
+            .model()
+            .await
+            .inspect_err(|e| log::error!("Failed to get model for webhook {channel_id}: {e:?}"))
+            .map_err(|_| DatabaseOrDiscordError::DiscordDeserializationFailed)?
+    };
+
+    Ok(webhook)
+}
+
 impl Channel {
     pub async fn get_by_id(id: u64, session: &Session) -> Result<Option<Self>, DatabaseError> {
         let id = u64_to_i64(id);
@@ -520,15 +554,7 @@ impl Channel {
         {
             Ok(channel)
         } else {
-            let webhook = http
-                .create_webhook(Id::new(id), "Payphone Messages")
-                .await
-                .inspect_err(|e| log::error!("Failed to create webhook for channel {id}: {e:?}"))
-                .map_err(DatabaseOrDiscordError::Discord)?
-                .model()
-                .await
-                .inspect_err(|e| log::error!("Failed to get model for webhook {id}: {e:?}"))
-                .map_err(|_| DatabaseOrDiscordError::DiscordDeserializationFailed)?;
+            let webhook = get_or_create_channel_webhook(id, http).await?;
 
             let response = http
                 .channel(Id::new(id))
@@ -593,6 +619,21 @@ impl Channel {
 
         Ok(())
     }
+
+    pub async fn update_webhook(
+        &mut self,
+        session: &Session,
+        http: &Client,
+    ) -> Result<(), DatabaseOrDiscordError> {
+        let webhook = get_or_create_channel_webhook(self.id, http).await?;
+
+        self.webhook_id = Some(webhook.id.get());
+        self.webhook_token = webhook.token;
+
+        self.upsert(session).await?;
+
+        Ok(())
+    }
 }
 
 #[derive(Debug, Clone)]

src/groups/calls/events.rs

@@ -2,6 +2,7 @@ use std::{error::Error, sync::Arc};
 
 use async_trait::async_trait;
 use itertools::Itertools;
+use twilight_http::{api_error::ApiError, error::ErrorType};
 use twilight_model::{
     gateway::payload::incoming::{MessageCreate, MessageDelete, MessageUpdate},
     id::Id,
@@ -151,7 +152,7 @@ impl OnMessageHandler<State> for CallMessageHandler {
             return Ok(());
         }
 
-        let Some(opposite_channel) = Channel::get_by_id(
+        let Some(mut opposite_channel) = Channel::get_by_id(
             call.other_party_channel_id(event.channel_id.get()),
             &rt.state.db,
         )
@@ -167,7 +168,7 @@ impl OnMessageHandler<State> for CallMessageHandler {
 
         let Some((webhook_id, webhook_token)) = opposite_channel
             .webhook_id
-            .zip(opposite_channel.webhook_token)
+            .zip(opposite_channel.webhook_token.clone())
         else {
             // Handle DM calls in the future.
             return Ok(());
@@ -202,14 +203,14 @@ impl OnMessageHandler<State> for CallMessageHandler {
             rt.http
                 .create_message(event.channel_id)
                 .content(
-                    "❌ Your message has been blocked because it broke Payphone's code of conduct. Check your DMs for more information.",
+                    "🚩 Your message has been blocked because it broke Payphone's code of conduct. Check your DMs for more information.",
                 )
                 .await?;
 
             rt.http
                 .create_message(Id::new(opposite_channel.id))
                 .content(&format!(
-                    "❌ **{}** Sent a message that was blocked.\n-# If you see an inappropiate message, please report it using `Message > Apps > Report Message`.",
+                    "🚩 **{}** Sent a message that was blocked.\n-# If you see an inappropiate message, please report it using `Message > Apps > Report Message`.",
                     profile.name.sanitize(false)
                 ))
                 .await?;
@@ -241,7 +242,7 @@ impl OnMessageHandler<State> for CallMessageHandler {
                 );
             }
 
-            let message_sent = rt
+            let message_sent = match rt
                 .http
                 .execute_webhook(Id::new(webhook_id), &webhook_token)
                 .allowed_mentions(None)
@@ -257,9 +258,26 @@ impl OnMessageHandler<State> for CallMessageHandler {
                         call.id,
                         opposite_channel.id
                     );
-                })?
-                .model()
-                .await?;
+                }) {
+                Ok(message) => message,
+                Err(e) => {
+                    if let ErrorType::Response {
+                        error: ApiError::General(error),
+                        ..
+                    } = e.kind()
+                    {
+                        if error.code == 10015 {
+                            log::debug!("Webhook not found, updating...");
+                            log::trace!("Webhook ID before the update: {}", opposite_channel.webhook_id.unwrap_or(0));
+                            opposite_channel.update_webhook(&rt.state.db, &rt.http).await?;
+                            log::trace!("Webhook ID after the update: {}", opposite_channel.webhook_id.unwrap_or(0));
+                        }
+                    }
+                    return Ok(());
+                }
+            }
+            .model()
+            .await?;
 
             message = Message::build()
                 .call_id(call.id)
@@ -306,6 +324,13 @@ pub struct CallMessageEditHandler;
 #[async_trait]
 impl OnMessageEditHandler<State> for CallMessageEditHandler {
     async fn on_message_edit(&self, event: &MessageUpdate, rt: Arc<Runtime<State>>) -> EventResult {
+        log::trace!(
+            "Edited message was '{}' sent by {} (bot? {:?})",
+            event.content,
+            event.author.id.get(),
+            event.author.bot
+        );
+
         let Some(channel) = Channel::get_by_id(event.channel_id.get(), &rt.state.db).await? else {
             return Ok(());
         };
@@ -346,7 +371,7 @@ impl OnMessageEditHandler<State> for CallMessageEditHandler {
                 rt.http
                     .create_message(event.channel_id)
                     .content(
-                        "❌ Your message's edit has been blocked because it broke Payphone's code of conduct. Check your DMs for more information.",
+                        "🚩 Your message's edit has been blocked because it broke Payphone's code of conduct. Check your DMs for more information.",
                     )
                     .reply(event.id)
                     .await?;

src/groups/calls/lists/explicit-sexual-language.txt

@@ -1,5 +1,7 @@
 sex
+segs
 esex
+esegs
 give head
 gave head
 give me head
@@ -35,7 +37,11 @@ stick it in yr pussy
 cock
 dick
 dih
-pussy
+pusy
+posay
+vagina
+vgina
+vgna
 dck
 psy
 pu-c
@@ -92,6 +98,7 @@ masturbate
 masturbating
 masterbating
 masterbaiting
+masturbatigng
 horny
 hrny
 whore

src/groups/calls/lists/hate-speech-and-slurs.txt

@@ -8,6 +8,7 @@ qiga
 niqa
 nqa
 nlger
+ngiga
 digga
 fag
 fagot
@@ -45,4 +46,5 @@ get the fvck out
 get the fck out
 get tfo
 blackie
-motherfucker
+motherfucker
+chinese ape

src/groups/calls/lists/threats-and-self-harm.txt

@@ -12,6 +12,3 @@ rape u
 slit
 grape
 graped
-throat slit
-slit your throat
-slit ur throat

src/groups/leveling/events.rs

@@ -0,0 +1,80 @@
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use deadpool_redis::redis::{AsyncTypedCommands, IntegerReplyOrNoOp};
+use rand::{Rng, SeedableRng};
+use rand_chacha::ChaCha20Rng;
+use time::{Date, Month, OffsetDateTime, Time};
+use twilight_model::gateway::payload::incoming::MessageCreate;
+
+use crate::{
+    database::models::Channel,
+    framework::{
+        events::traits::{EventResult, OnMessageHandler},
+        runtime::Runtime,
+    },
+    state::State,
+};
+
+const KEY_LEADERBOARD_MONTHLY: &str = "lvl:monthly";
+
+fn seconds_until_end_of_month() -> i64 {
+    let now = OffsetDateTime::now_utc();
+    let year = now.year();
+    let month = now.month();
+
+    // figure out the first day of the next month
+    let (next_year, next_month) = if month == Month::December {
+        (year + 1, Month::January)
+    } else {
+        (year, month.next())
+    };
+
+    let first_of_next_month = Date::from_calendar_date(next_year, next_month, 1)
+        .unwrap()
+        .with_time(Time::MIDNIGHT)
+        .assume_utc();
+
+    (first_of_next_month - now).whole_seconds()
+}
+
+#[derive(Debug)]
+pub struct MessageLevelingHandler;
+
+#[async_trait]
+impl OnMessageHandler<State> for MessageLevelingHandler {
+    async fn on_message(&self, event: &MessageCreate, rt: Arc<Runtime<State>>) -> EventResult {
+        let Some(channel) = Channel::get_by_id(event.channel_id.get(), &rt.state.db).await? else {
+            return Ok(());
+        };
+
+        if channel.call_id.is_none() {
+            return Ok(());
+        }
+
+        let mut redis = rt.state.redis.get().await?;
+
+        let key = format!("lvl:{}", event.author.id);
+
+        let lock = redis.exists(&key).await?;
+
+        if lock {
+            return Ok(());
+        }
+
+        let xp = ChaCha20Rng::from_os_rng().random_range(10..=25);
+
+        redis.set_ex(&key, true, 60).await?;
+        redis
+            .zincr(KEY_LEADERBOARD_MONTHLY, event.author.id.to_string(), xp)
+            .await?;
+
+        let leaderboard_exat = redis.ttl(KEY_LEADERBOARD_MONTHLY).await?;
+
+        if leaderboard_exat == IntegerReplyOrNoOp::ExistsButNotRelevant {
+            redis.expire(key, seconds_until_end_of_month()).await?;
+        }
+
+        Ok(())
+    }
+}

src/groups/leveling/mod.rs

@@ -0,0 +1 @@
+pub mod events;

src/groups/mod.rs

@@ -17,5 +17,6 @@
 pub mod about;
 pub mod calls;
 pub mod gateway;
-pub mod servers;
+pub mod leveling;
 pub mod moderation;
+pub mod servers;

src/main.rs

@@ -125,16 +125,13 @@ use crate::{
         runtime::Runtime,
     },
     groups::{
-        about::{commands::HelpCommand, components::HelpGoBackButton},
-        calls::{
+        about::{commands::HelpCommand, components::HelpGoBackButton}, calls::{
             commands::{CallCommand, FriendRequestCommand, HangupCommand, SkipCommand},
             events::{CallMessageDeleteHandler, CallMessageEditHandler, CallMessageHandler},
-        },
-        gateway::events::{ReadyHandler, ReconnectHandler},
-        servers::{
+        }, gateway::events::{ReadyHandler, ReconnectHandler}, leveling::events::MessageLevelingHandler, servers::{
             commands::PrefixCommand,
             events::{ServerCreateHandler, ServerDeleteHandler, ServerUpdateHandler},
-        },
+        }
     },
     state::State,
 };
@@ -146,6 +143,7 @@ mod discord;
 mod framework;
 mod groups;
 mod logging;
+mod redis;
 mod state;
 mod utils;
 
@@ -169,6 +167,7 @@ pub async fn main() {
                 database_username,
                 database_password,
                 shard,
+                redis_urls,
             } => {
                 start(
                     token,
@@ -176,6 +175,7 @@ pub async fn main() {
                     database_username,
                     database_password,
                     shard.into(),
+                    redis_urls,
                 )
                 .await;
             }
@@ -217,6 +217,7 @@ async fn start(
     database_username: Option<String>,
     database_password: Option<String>,
     shard: ShardId,
+    redis_urls: Vec<String>,
 ) {
     log::info!("Welcome to Payphone, Nyeki!");
     log::info!(
@@ -230,7 +231,14 @@ async fn start(
 
     log::info!("Database connection established successfully.");
 
-    let state = State { db };
+    let redis_pool = redis::init(redis_urls).expect("Could not connect to Redis!");
+
+    log::info!("Redis connection established successfully.");
+
+    let state = State {
+        db,
+        redis: redis_pool,
+    };
 
     let bot = Bot::new(state, ServerPrefixOrPayphoneMention)
         .shard(shard)
@@ -252,6 +260,7 @@ async fn start(
         .on_ready(ReadyHandler)
         .on_reconnect(ReconnectHandler)
         .on_message(CallMessageHandler)
+        .on_message(MessageLevelingHandler)
         .on_message_edit(CallMessageEditHandler)
         .on_message_delete(CallMessageDeleteHandler)
         .on_server_create(ServerCreateHandler)

src/redis.rs

@@ -0,0 +1,8 @@
+use deadpool_redis::{
+    Runtime,
+    cluster::{Config, CreatePoolError, Pool},
+};
+
+pub fn init(urls: Vec<String>) -> Result<Pool, CreatePoolError> {
+    Config::from_urls(urls).create_pool(Some(Runtime::Tokio1))
+}

src/state.rs

@@ -18,8 +18,10 @@
 //! }
 //! ```
 
+use deadpool_redis::cluster::Pool;
 use scylla::client::session::Session;
 
 pub struct State {
     pub db: Session,
+    pub redis: Pool,
 }