use std::collections::HashSet; use scylla::{ DeserializeValue, SerializeValue, client::session::Session, deserialize::row::DeserializeRow, errors::{ DeserializationError, ExecutionError, FirstRowError, IntoRowsResultError, RowsError, SingleRowError, }, response::query_result::QueryRowsResult, statement::{ Statement, batch::{Batch, BatchType}, }, }; use thiserror::Error; use time::{Duration, OffsetDateTime, UtcDateTime, UtcOffset}; use tokio::sync::mpsc::Sender; use twilight_http::Client; use twilight_model::{ channel::{Message as TwilightMessage, Webhook}, guild::Guild as TwilightGuild, id::Id, user::User as TwilightUser, }; use twilight_validate::request::{ValidationError, ValidationErrorType, webhook_username}; use crate::{ colors::AvatarColorFetchingRequest, constants::DEFAULT_PREFIX, groups::calls::masks::{get_mask_avatar_url, get_mask_display_name, get_random_mask}, utils::{Bye, get_random_u64, i64_to_u64, u64_to_i64}, }; #[derive(Debug, Error)] pub enum DatabaseOrDiscordError { #[error("An error occurred when querying the database")] Database(#[from] DatabaseError), #[error("An error occurred when querying the Discord API")] Discord(#[from] twilight_http::Error), #[error("The Discord API returned a response that could not be deserialized")] DiscordDeserializationFailed, } #[derive(Debug, Error)] pub enum DatabaseError { #[error("An error occurred when querying the database")] QueryError(#[from] ExecutionError), #[error("The database responded but the result of the query weren't rows")] ResultNotRows(#[from] IntoRowsResultError), #[error("The resulting row could not be deserialized into an object")] DeserializationFailed(#[from] DeserializationError), #[error("The database responded but the result of the query weren't rows")] RowsError(#[from] RowsError), #[error("The resulting row could not be deserialized into an object")] SingleRowFailed(#[from] SingleRowError), #[error("The database responded but the first row raised an error")] FirstRowError(#[from] FirstRowError), #[error("An unknown error occurred")] Unknown(&'static str), } trait Object<'a> { type Row: DeserializeRow<'a, 'a>; fn from_row(row: Self::Row) -> Self; fn to_row(value: Self) -> Self::Row; } /// Returns the first row of the result set, if any. /// /// Arguments: /// * `result` - The DB result. /// /// Returns: /// [`Result, DatabaseError>`] - The first row of the result set, if any. async fn get_single_row_from_result<'a, T: Object<'a>>( result: &'a QueryRowsResult, ) -> Result, DatabaseError> { match result.single_row::() { Ok(row) => Ok(Some(T::from_row(row))), Err(error) => match error { SingleRowError::UnexpectedRowCount(_) => Ok(None), _ => Err(error.into()), }, } } #[derive(Debug, Clone)] pub struct User { pub id: u64, pub name: String, pub email: Option, pub is_masked: bool, pub is_moderator: bool, pub joined_at: UtcDateTime, pub last_seen_at: UtcDateTime, pub supporter_since: Option, pub supporter_until: Option, pub is_banned_from_leaderboard: Option, pub findings: HashSet, pub avatar_hash: Option, pub color: Option, } impl<'a> Object<'a> for User { type Row = ( i64, String, Option, bool, bool, OffsetDateTime, OffsetDateTime, Option, Option, Option, HashSet, Option, Option, Option, ); fn from_row(row: Self::Row) -> Self { Self { id: i64_to_u64(row.0), name: row.1, email: row.2, is_masked: row.3, is_moderator: row.4, joined_at: row.5.into(), last_seen_at: row.6.into(), supporter_since: row.7.map(|i| i.into()), supporter_until: row.8.map(|i| i.into()), is_banned_from_leaderboard: row.9, findings: row.10, avatar_hash: row.11, color: row.12, } } fn to_row(value: Self) -> Self::Row { ( u64_to_i64(value.id), value.name, value.email, value.is_masked, value.is_moderator, value.joined_at.into(), value.last_seen_at.into(), value.supporter_since.map(|i| i.into()), value.supporter_until.map(|i| i.into()), Some(value.is_banned_from_leaderboard.unwrap_or(false)), value.findings, value.avatar_hash, value.color, Some(value.color.is_some()), ) } } pub struct NewUserIncident { pub counting_incidents: usize, pub is_first_incident: bool, } impl User { pub async fn get_by_id(id: u64, session: &Session) -> Result, DatabaseError> { let id = u64_to_i64(id); let result = session .query_unpaged( r#" SELECT user_id, name, email, is_masked, is_moderator, joined_at, last_seen_at, supporter_since, supporter_until, is_banned_from_leaderboard, findings, avatar_hash, color, is_color_set FROM payphone.users WHERE user_id = ? LIMIT 1; "#, (id,), ) .await? .into_rows_result()?; get_single_row_from_result(&result).await } pub async fn get_by_name( name: String, session: &Session, ) -> Result, DatabaseError> { let Some(user_by_name) = UserByName::get(name, session).await? else { return Ok(None); }; Self::get_by_id(user_by_name.user_id, session).await } /// Gets a user from the DB and sets their last seen timestamp or inserts a new user from Discord. /// /// It also updates the user's username. /// /// Arguments: /// * `user` - The Twilight user to get or create. /// * `session` - The Scylla session to use for database operations. /// /// Returns: /// [`Result`] - The user from the database or a new user created from /// Discord. pub async fn create_or_update( user: TwilightUser, session: &Session, sender: &Sender, ) -> Result { let db_user = User::get_by_id(user.id.get(), session).await?; if let Some(mut db_user) = db_user { db_user.last_seen_at = UtcDateTime::now(); db_user.name = user.name; if user.avatar.map(|hash| hash.to_string()) != db_user.avatar_hash { db_user.avatar_hash = user.avatar.map(|hash| hash.to_string()); db_user.color = None; if let Some(avatar_hash) = &db_user.avatar_hash { sender .send(AvatarColorFetchingRequest::new(avatar_hash, user.id.get())) .await .inspect_err(|e| { log::error!( "Failed to send avatar color fetching request to queue! {e}" ) }) .bye(); } } return Ok(db_user); } let new_user = User::from(user); if let Some(avatar_hash) = &new_user.avatar_hash { sender .send(AvatarColorFetchingRequest::new(avatar_hash, new_user.id)) .await .inspect_err(|e| { log::error!("Failed to send avatar color fetching request to queue! {e}") }) .bye(); } new_user.insert(session).await?; Ok(new_user) } pub async fn insert(&self, session: &Session) -> Result<(), DatabaseError> { session .query_unpaged( r#" INSERT INTO payphone.users ( user_id, name, email, is_masked, is_moderator, joined_at, last_seen_at, supporter_since, supporter_until, is_banned_from_leaderboard, findings, avatar_hash, color, is_color_set ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?); "#, Self::to_row(self.clone()), ) .await .map_err(DatabaseError::QueryError)?; Ok(()) } pub async fn incident( &self, reason: String, message: Option, created_by_id: Option, session: &Session, ) -> Result { let incidents = UserIncident::get_for_user(self.id, session).await?; let counting_incidents = incidents .iter() .filter(|incident| incident.created_at + Duration::weeks(12) > UtcDateTime::now()) .count(); let incident_type = if counting_incidents != 0 { UserIncidentType::Ban } else { UserIncidentType::Warning }; let duration = match counting_incidents { 0 => None, // Warning 1 => Some(Duration::days(1)), // First ban _ => None, // Permanent ban }; let incident = UserIncident { user_id: self.id, incident_id: get_random_u64(), report_id: None, r#type: incident_type, created_at: UtcDateTime::now(), expires_at: duration.map(|d| UtcDateTime::now() + d), message, reason, created_by_id, duration: duration.map(|_| "1 day".into()), }; incident.insert(session).await?; Ok(NewUserIncident { counting_incidents, is_first_incident: incidents.is_empty(), }) } pub async fn count(session: &Session) -> Result { let result = session .query_unpaged( r#" SELECT user_id FROM payphone.users; "#, (), ) .await? .into_rows_result()?; Ok(result.rows_num()) } } impl From for User { fn from(value: TwilightUser) -> Self { Self { id: value.id.get(), name: value.name, email: value.email, is_masked: false, is_moderator: false, joined_at: UtcDateTime::now(), last_seen_at: UtcDateTime::now(), supporter_since: None, supporter_until: None, is_banned_from_leaderboard: None, findings: HashSet::new(), avatar_hash: value.avatar.map(|a| a.to_string()), color: None, } } } #[derive(Debug, Clone)] struct UserByName { pub user_id: u64, pub name: String, } impl<'a> Object<'a> for UserByName { type Row = (i64, String); fn from_row(row: Self::Row) -> Self { Self { user_id: i64_to_u64(row.0), name: row.1, } } fn to_row(value: Self) -> Self::Row { (u64_to_i64(value.user_id), value.name) } } impl UserByName { async fn get(name: String, session: &Session) -> Result, DatabaseError> { let result = session .query_unpaged( r#" SELECT user_id, name FROM payphone.users_by_name WHERE name = ? LIMIT 1; "#, (name,), ) .await? .into_rows_result()?; get_single_row_from_result(&result).await } } #[derive(Debug, Clone)] pub struct Server { pub id: u64, pub name: String, pub prefix: String, pub language: String, } impl<'a> Object<'a> for Server { type Row = (i64, String, String, String); fn from_row(row: Self::Row) -> Self { Self { id: i64_to_u64(row.0), name: row.1, prefix: row.2, language: row.3, } } fn to_row(value: Self) -> Self::Row { ( u64_to_i64(value.id), value.name, value.prefix, value.language, ) } } impl Server { pub async fn get_by_id(id: u64, session: &Session) -> Result, DatabaseError> { let id = u64_to_i64(id); let result = session .query_unpaged( r#" SELECT server_id, name, prefix, language FROM payphone.servers WHERE server_id = ? LIMIT 1; "#, (id,), ) .await? .into_rows_result()?; get_single_row_from_result(&result).await } pub async fn get_or_create_by_id( id: u64, session: &Session, http: &Client, ) -> Result { let server = Server::get_by_id(id, session) .await .map_err(DatabaseOrDiscordError::Database)?; if let Some(server) = server { Ok(server) } else { let server: Server = http .guild(Id::new(id)) .await .map_err(DatabaseOrDiscordError::Discord)? .model() .await .map_err(|_| DatabaseOrDiscordError::DiscordDeserializationFailed)? .into(); server .insert(session) .await .map_err(DatabaseOrDiscordError::Database)?; Ok(server) } } pub async fn insert(&self, session: &Session) -> Result<(), DatabaseError> { session .query_unpaged( r#" INSERT INTO payphone.servers ( server_id, name, prefix, language ) VALUES (?, ?, ?, ?) IF NOT EXISTS; "#, Self::to_row(self.clone()), ) .await .inspect_err(|e| log::error!("{e:?}")) .map_err(DatabaseError::QueryError)?; Ok(()) } pub async fn upsert(&self, session: &Session) -> Result<(), DatabaseError> { session .query_unpaged( r#" INSERT INTO payphone.servers ( server_id, name, prefix, language ) VALUES ( ?, ?, ?, ? ); "#, Self::to_row(self.clone()), ) .await .map_err(DatabaseError::QueryError)?; Ok(()) } pub async fn delete_by_id(id: u64, session: &Session) -> Result<(), DatabaseError> { session .query_unpaged( "DELETE FROM payphone.servers WHERE server_id = ?;", (u64_to_i64(id),), ) .await .map_err(DatabaseError::QueryError)?; Ok(()) } } impl From for Server { fn from(value: TwilightGuild) -> Self { Self { id: value.id.get(), name: value.name, prefix: DEFAULT_PREFIX.into(), language: value.preferred_locale, } } } #[derive(Debug, Clone)] pub struct Channel { pub id: u64, pub server_id: u64, pub call_id: Option, pub is_searching: bool, pub recently_called_server_ids: Vec, pub webhook_id: Option, pub webhook_token: Option, } impl<'a> Object<'a> for Channel { type Row = ( i64, i64, Option, bool, Vec, Option, Option, ); fn from_row(row: Self::Row) -> Self { Self { id: i64_to_u64(row.0), server_id: i64_to_u64(row.1), call_id: row.2.map(i64_to_u64), is_searching: row.3, recently_called_server_ids: row.4.iter().map(|&i| i64_to_u64(i)).collect(), webhook_id: row.5.map(i64_to_u64), webhook_token: row.6, } } fn to_row(value: Self) -> Self::Row { ( u64_to_i64(value.id), u64_to_i64(value.server_id), value.call_id.map(u64_to_i64), value.is_searching, value .recently_called_server_ids .iter() .map(|&u| u64_to_i64(u)) .collect(), value.webhook_id.map(u64_to_i64), value.webhook_token, ) } } async fn get_or_create_channel_webhook( channel_id: u64, http: &Client, ) -> Result { 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, DatabaseError> { let id = u64_to_i64(id); let result = session .query_unpaged( r#" SELECT channel_id, server_id, call_id, is_searching, recently_called_server_ids, webhook_id, webhook_token FROM payphone.channels WHERE channel_id = ? LIMIT 1; "#, (id,), ) .await? .into_rows_result()?; get_single_row_from_result(&result).await } pub async fn get_or_create_by_id( id: u64, session: &Session, http: &Client, ) -> Result { if let Some(channel) = Channel::get_by_id(id, session) .await .map_err(DatabaseOrDiscordError::Database)? { Ok(channel) } else { let webhook = get_or_create_channel_webhook(id, http).await?; let response = http .channel(Id::new(id)) .await .map_err(DatabaseOrDiscordError::Discord)? .model() .await .inspect_err(|e| { log::error!("The discord API didn't return a parseable channel! {e:?}") }) .map_err(|_| DatabaseOrDiscordError::DiscordDeserializationFailed)?; let channel = Channel { id: response.id.get(), server_id: response.guild_id.map(|i| i.get()).unwrap_or(0), call_id: None, is_searching: false, recently_called_server_ids: vec![], webhook_id: Some(webhook.id.get()), webhook_token: webhook.token, }; channel .upsert(session) .await .inspect_err(|e| log::error!("Could not upsert channel in database! {e:?}")) .map_err(DatabaseOrDiscordError::Database)?; Ok(channel) } } pub async fn upsert(&self, session: &Session) -> Result<(), DatabaseError> { session .query_unpaged( r#" INSERT INTO payphone.channels ( channel_id, server_id, call_id, is_searching, recently_called_server_ids, webhook_id, webhook_token ) VALUES (?, ?, ?, ?, ?, ?, ?); "#, Self::to_row(self.clone()), ) .await .map_err(DatabaseError::QueryError)?; Ok(()) } pub async fn cancel_search(&self, session: &Session) -> Result<(), DatabaseError> { session .query_unpaged( "UPDATE payphone.channels SET is_searching = false WHERE channel_id = ?;", (u64_to_i64(self.id),), ) .await .map_err(DatabaseError::QueryError)?; 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)] pub struct SearchingChannel { pub id: u64, pub server_id: u64, pub recently_called_server_ids: Vec, } impl<'a> Object<'a> for SearchingChannel { type Row = (i64, i64, Vec); fn from_row(row: Self::Row) -> Self { Self { id: i64_to_u64(row.0), server_id: i64_to_u64(row.1), recently_called_server_ids: row.2.iter().map(|&i| i64_to_u64(i)).collect(), } } fn to_row(value: Self) -> Self::Row { ( u64_to_i64(value.id), u64_to_i64(value.server_id), value .recently_called_server_ids .iter() .map(|&u| u64_to_i64(u)) .collect(), ) } } impl SearchingChannel { pub async fn get_all(session: &Session) -> Result, DatabaseError> { let result = session .query_unpaged( r#" SELECT channel_id, server_id, recently_called_server_ids FROM payphone.searching_channels; "#, (), ) .await? .into_rows_result()?; Ok(result .rows::<::Row>()? .collect::, _>>()? .into_iter() .map(Self::from_row) .collect::>()) } } #[derive(Debug, Clone)] pub struct Call { pub id: u64, pub channel_ids: Vec, pub started_at: UtcDateTime, pub finished_at: Option, } impl<'a> Object<'a> for Call { type Row = (i64, Vec, OffsetDateTime, Option); fn from_row(row: Self::Row) -> Self { Self { id: i64_to_u64(row.0), channel_ids: row.1.iter().map(|&i| i64_to_u64(i)).collect(), started_at: row.2.into(), finished_at: row.3.map(|i| i.into()), } } fn to_row(value: Self) -> Self::Row { ( u64_to_i64(value.id), value.channel_ids.iter().map(|&u| u64_to_i64(u)).collect(), value.started_at.into(), value.finished_at.map(|i| i.into()), ) } } fn push_and_cap(vec: &mut Vec, value: T, max_size: usize) { if vec.len() >= max_size { vec.remove(0); } vec.push(value); } impl Call { pub async fn get_by_id(id: u64, session: &Session) -> Result, DatabaseError> { let id = u64_to_i64(id); let result = session .query_unpaged( r#" SELECT call_id, channel_ids, started_at, finished_at FROM payphone.calls WHERE call_id = ? LIMIT 1; "#, (id,), ) .await .map_err(DatabaseError::QueryError)? .into_rows_result()?; get_single_row_from_result(&result).await } /// Returns the opposite party's channel ID. pub fn other_party_channel_id(&self, channel_id: u64) -> u64 { if self.channel_ids[0] == channel_id { self.channel_ids[1] } else { self.channel_ids[0] } } pub fn is_active(&self) -> bool { self.finished_at.is_none() } /// Starts a call, or marks the channel as searching if the call is not started. /// /// Arguments: /// * `channel` - The channel to start the call in. /// * `session` - The Scylla session to use for database operations. /// /// Returns: /// [`Result, DatabaseError>`] - `Ok(Some(Call))` if the call was started /// successfully, or `Ok(None)` if the channel was marked as searching. If an error occurs, it /// returns `Err(DatabaseError)`. pub async fn start( channel: &Channel, session: &Session, ) -> Result, DatabaseError> { let searching_channels = SearchingChannel::get_all(session).await?; let call_id = get_random_u64(); log::trace!("Call ID is {call_id}"); for searching_channel in searching_channels .iter() .filter(|sc| !sc.recently_called_server_ids.contains(&channel.server_id)) { if searching_channel.server_id == channel.server_id { return Ok(None); } if channel .recently_called_server_ids .contains(&searching_channel.server_id) { continue; } log::trace!("Found a searching channel: {searching_channel:?}"); let mut searching_channel_recently_called_server_ids = searching_channel .recently_called_server_ids .clone() .into_iter() .map(u64_to_i64) .collect::>(); push_and_cap( &mut searching_channel_recently_called_server_ids, u64_to_i64(channel.server_id), 5, ); let (is_applied, ..) = session .query_unpaged( r#" UPDATE payphone.channels SET is_searching = false, call_id = ?, recently_called_server_ids = ? WHERE channel_id = ? IF call_id = NULL; "#, ( u64_to_i64(call_id), searching_channel_recently_called_server_ids, u64_to_i64(searching_channel.id), ), ) .await .inspect_err(|e| log::error!("Failed to add channel to the call! {e:?}"))? .into_rows_result() .inspect_err(|e| log::error!("Failed to get result of the query! {e:?}"))? .first_row::<(bool, Option)>() .inspect_err(|e| log::error!("Failed to get first row of the result! {e:?}"))?; log::trace!("Is applied: {is_applied:?}"); if is_applied { let mut channel_recently_called_server_ids = channel .recently_called_server_ids .clone() .into_iter() .map(u64_to_i64) .collect::>(); push_and_cap( &mut channel_recently_called_server_ids, u64_to_i64(searching_channel.server_id), 5, ); session .query_unpaged( "UPDATE payphone.channels SET call_id = ?, recently_called_server_ids = ? WHERE channel_id = ?", (u64_to_i64(call_id), channel_recently_called_server_ids, u64_to_i64(channel.id)), ) .await .inspect_err(|e| log::error!("Failed to update channel {e:?}")) .map_err(DatabaseError::QueryError)?; let call = Call { id: call_id, channel_ids: vec![searching_channel.id, channel.id].into_iter().collect(), started_at: UtcDateTime::now(), finished_at: None, }; call.insert(session) .await .inspect_err(|e| log::error!("Failed to insert call {e:?}"))?; return Ok(Some(call)); } else { // This channel already got a call, so we skip it, preventing race conditions. continue; } } session.query_unpaged( "UPDATE payphone.channels SET is_searching = true WHERE channel_id = ? IF call_id = NULL;", (u64_to_i64(channel.id),), ) .await .inspect_err(|e| log::error!("Failed to mark channel as searching! {e:?}")) .map_err(DatabaseError::QueryError)?; Ok(None) } pub async fn insert(&self, session: &Session) -> Result<(), DatabaseError> { session .query_unpaged( r#" INSERT INTO payphone.calls ( call_id, channel_ids, started_at, finished_at ) VALUES (?, ?, ?, ?); "#, Self::to_row(self.clone()), ) .await .map_err(DatabaseError::QueryError)?; Ok(()) } pub async fn hangup(&self, session: &Session) -> Result<(), DatabaseError> { let mut batch = Batch::new(BatchType::Unlogged); batch.append_statement(Statement::new( "UPDATE payphone.calls SET finished_at = ? WHERE call_id = ?;", )); batch.append_statement(Statement::new( "UPDATE payphone.channels SET call_id = NULL WHERE channel_id = ?;", )); batch.append_statement(Statement::new( "UPDATE payphone.channels SET call_id = NULL WHERE channel_id = ?;", )); let call_id = u64_to_i64(self.id); session .batch( &batch, ( (OffsetDateTime::now_utc(), call_id), (u64_to_i64(self.channel_ids[0]),), (u64_to_i64(self.channel_ids[1]),), ), ) .await .map_err(DatabaseError::QueryError)?; log::trace!("Hung up call {call_id}"); Ok(()) } pub async fn count(session: &Session) -> Result { let result = session .query_unpaged( r#" SELECT call_id FROM payphone.calls; "#, (), ) .await? .into_rows_result()?; Ok(result.rows_num()) } } #[derive(Debug, Clone, SerializeValue, DeserializeValue)] pub struct MessageData { pub id: i64, pub call_id: i64, pub author_id: i64, pub author_name: String, pub author_avatar_url: String, pub author_mask: Option, pub content: String, pub attachments: Vec, pub sticker: Option, pub propagation_message_id: Option, pub propagation_channel_id: Option, pub replying_to_message_id: Option, pub is_deleted: bool, pub is_edited: bool, pub is_censored: bool, } #[derive(Debug, Error)] #[error("The message was missing either the event's message, the user's profile, or the call ID")] pub struct MessageBuilderError; #[derive(Default)] pub struct MessageBuilder { message: Option, participant: Option, call_id: Option, is_censored: bool, propagation_message_id: Option, propagation_channel_id: Option, replying_to_message_id: Option, } impl MessageBuilder { fn new() -> Self { Self { message: None, participant: None, call_id: None, is_censored: false, propagation_message_id: None, propagation_channel_id: None, replying_to_message_id: None, } } pub fn message(mut self, message: TwilightMessage) -> Self { self.message = Some(message); self } pub fn participant(mut self, participant: Participant) -> Self { self.participant = Some(participant); self } pub fn call_id(mut self, call_id: u64) -> Self { self.call_id = Some(call_id); self } #[allow(clippy::wrong_self_convention)] pub fn is_censored(mut self, is_censored: bool) -> Self { self.is_censored = is_censored; self } pub fn propagation_message_id(mut self, propagation_message_id: u64) -> Self { self.propagation_message_id = Some(propagation_message_id); self } pub fn propagation_channel_id(mut self, propagation_channel_id: u64) -> Self { self.propagation_channel_id = Some(propagation_channel_id); self } // Argument is made optional here because it happens to be more convenient for where this is // used. pub fn replying_to_message_id(mut self, replying_to_id: Option) -> Self { self.replying_to_message_id = replying_to_id; self } pub fn build(self) -> Result { let message = self.message.ok_or(MessageBuilderError)?; let participant = self.participant.ok_or(MessageBuilderError)?; let call_id = self.call_id.ok_or(MessageBuilderError)?; let data = MessageData { id: u64_to_i64(message.id.get()), call_id: u64_to_i64(call_id), author_id: u64_to_i64(participant.user_id), author_name: participant.user_name, author_avatar_url: participant.user_avatar_url, author_mask: participant.is_masked.then_some(participant.user_mask), content: message.content, attachments: message.attachments.iter().map(|a| a.url.clone()).collect(), sticker: message .sticker_items .first() .map(|s| u64_to_i64(s.id.get())), propagation_message_id: self.propagation_message_id.map(u64_to_i64), propagation_channel_id: self.propagation_channel_id.map(u64_to_i64), replying_to_message_id: self.replying_to_message_id.map(u64_to_i64), // You can't create a deleted message, this is always going to be false when // initializing is_deleted: false, is_edited: message.edited_timestamp.is_some(), is_censored: self.is_censored, }; Ok(Message { call_id, message_id: message.id.get(), author_id: participant.user_id, propagation_message_id: self.propagation_message_id, data, }) } } #[derive(Debug, Clone)] pub struct Message { pub call_id: u64, pub message_id: u64, pub author_id: u64, pub propagation_message_id: Option, pub data: MessageData, } impl<'a> Object<'a> for Message { type Row = (i64, i64, i64, Option, MessageData); fn from_row(row: Self::Row) -> Self { Self { call_id: i64_to_u64(row.0), message_id: i64_to_u64(row.1), author_id: i64_to_u64(row.2), propagation_message_id: row.3.map(i64_to_u64), data: row.4, } } fn to_row(value: Self) -> Self::Row { ( u64_to_i64(value.call_id), u64_to_i64(value.message_id), u64_to_i64(value.author_id), value.propagation_message_id.map(u64_to_i64), value.data, ) } } impl Message { pub fn build() -> MessageBuilder { MessageBuilder::new() } pub async fn get_by_call_and_message_id( call_id: u64, message_id: u64, session: &Session, ) -> Result, DatabaseError> { let call_id = u64_to_i64(call_id); let message_id_i64 = u64_to_i64(message_id); let result = session .query_unpaged( r#" SELECT call_id, message_id, author_id, propagation_message_id, data FROM payphone.messages WHERE call_id = ? AND message_id = ? LIMIT 1; "#, (call_id, message_id_i64), ) .await? .into_rows_result()?; if result.rows_num() == 0 { Self::get_by_propagation_message_id(message_id, session).await } else { get_single_row_from_result(&result).await } } pub async fn get_by_propagation_message_id( propagation_message_id: u64, session: &Session, ) -> Result, DatabaseError> { let message_by_propagation_id = MessageByPropagationId::get_by_propagation_message_id(propagation_message_id, session) .await?; if let Some(message_by_propagation_id) = message_by_propagation_id { let result = session .query_unpaged( r#" SELECT call_id, message_id, author_id, propagation_message_id, data FROM payphone.messages WHERE call_id = ? AND message_id = ? LIMIT 1; "#, ( u64_to_i64(message_by_propagation_id.call_id), u64_to_i64(message_by_propagation_id.message_id), ), ) .await? .into_rows_result()?; return get_single_row_from_result(&result).await; } Ok(None) } pub async fn get_last_by_author_id( call_id: u64, author_id: u64, session: &Session, ) -> Result, DatabaseError> { let message_by_author_id = MessageByAuthorId::get_last_by_author_id(call_id, author_id, session).await?; if let Some(message_by_author_id) = message_by_author_id { let result = session .query_unpaged( r#" SELECT call_id, message_id, author_id, propagation_message_id, data FROM payphone.messages WHERE call_id = ? AND message_id = ? LIMIT 1; "#, ( u64_to_i64(message_by_author_id.call_id), u64_to_i64(message_by_author_id.message_id), ), ) .await? .into_rows_result()?; return get_single_row_from_result(&result).await; } Ok(None) } pub async fn insert(&self, session: &Session) -> Result<(), DatabaseError> { session .query_unpaged( r#" INSERT INTO payphone.messages ( call_id, message_id, author_id, propagation_message_id, data ) VALUES (?, ?, ?, ?, ?); "#, Self::to_row(self.clone()), ) .await .map_err(DatabaseError::QueryError)?; Ok(()) } pub async fn count(session: &Session) -> Result { let result = session .query_unpaged( r#" SELECT message_id FROM payphone.messages; "#, (), ) .await? .into_rows_result()?; Ok(result.rows_num()) } } pub struct MessageByPropagationId { pub call_id: u64, pub message_id: u64, pub propagation_message_id: u64, } impl<'a> Object<'a> for MessageByPropagationId { type Row = (i64, i64, i64); fn from_row(row: Self::Row) -> Self { Self { call_id: i64_to_u64(row.0), message_id: i64_to_u64(row.1), propagation_message_id: i64_to_u64(row.2), } } fn to_row(value: Self) -> Self::Row { ( u64_to_i64(value.call_id), u64_to_i64(value.message_id), u64_to_i64(value.propagation_message_id), ) } } impl MessageByPropagationId { pub async fn get_by_propagation_message_id( propagation_message_id: u64, session: &Session, ) -> Result, DatabaseError> { let propagation_message_id = u64_to_i64(propagation_message_id); let result = session .query_unpaged( r#" SELECT call_id, message_id, propagation_message_id FROM payphone.messages_by_propagation_id WHERE propagation_message_id = ? LIMIT 1; "#, (propagation_message_id,), ) .await? .into_rows_result()?; get_single_row_from_result(&result).await } } pub struct MessageByAuthorId { pub call_id: u64, pub message_id: u64, pub author_id: u64, } impl<'a> Object<'a> for MessageByAuthorId { type Row = (i64, i64, i64); fn from_row(row: Self::Row) -> Self { Self { call_id: i64_to_u64(row.0), message_id: i64_to_u64(row.1), author_id: i64_to_u64(row.2), } } fn to_row(value: Self) -> Self::Row { ( u64_to_i64(value.call_id), u64_to_i64(value.message_id), u64_to_i64(value.author_id), ) } } impl MessageByAuthorId { pub async fn get_last_by_author_id( call_id: u64, author_id: u64, session: &Session, ) -> Result, DatabaseError> { let call_id = u64_to_i64(call_id); let author_id = u64_to_i64(author_id); let result = session .query_unpaged( r#" SELECT call_id, message_id, author_id FROM payphone.messages_by_author_id WHERE call_id = ? AND author_id = ? ORDER BY message_id DESC LIMIT 1; "#, (call_id, author_id), ) .await? .into_rows_result()?; get_single_row_from_result(&result).await } } #[derive(Debug, Clone, PartialEq, Eq)] pub enum ReportResolution { Ban, Warn, Skip, } #[derive(Debug, Clone)] pub struct Report { pub id: u64, pub user_id: u64, pub message: MessageData, pub replying_to: Option, pub created_at: UtcDateTime, pub reviewed_at: Option, pub created_by_ids: Vec, pub reviewed_by_id: Option, pub resolution: Option, } impl<'a> Object<'a> for Report { type Row = ( i64, i64, MessageData, Option, OffsetDateTime, Option, Vec, Option, Option, ); fn from_row(row: Self::Row) -> Self { Self { id: i64_to_u64(row.0), user_id: i64_to_u64(row.1), message: row.2, replying_to: row.3, created_at: row.4.to_utc(), reviewed_at: row.5.map(|i| i.to_utc()), created_by_ids: row.6.into_iter().map(i64_to_u64).collect(), reviewed_by_id: row.7.map(i64_to_u64), resolution: row.8.map(|r| match r.as_str() { "ban" => ReportResolution::Ban, "warn" => ReportResolution::Warn, "skip" => ReportResolution::Skip, _ => unreachable!(), }), } } fn to_row(value: Self) -> Self::Row { ( u64_to_i64(value.id), u64_to_i64(value.user_id), value.message, value.replying_to, value.created_at.to_offset(UtcOffset::UTC), value.reviewed_at.map(|i| i.to_offset(UtcOffset::UTC)), value.created_by_ids.into_iter().map(u64_to_i64).collect(), value.reviewed_by_id.map(u64_to_i64), value.resolution.map(|r| match r { ReportResolution::Ban => "ban".into(), ReportResolution::Warn => "warn".into(), ReportResolution::Skip => "skip".into(), }), ) } } impl Report { pub async fn insert(&self, session: &Session) -> Result<(), DatabaseError> { session .query_unpaged( r#" INSERT INTO payphone.reports ( report_id, user_id, message, replying_to, created_at, reviewed_at, created_by_ids, reviewed_by_id, resolution ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?); "#, Self::to_row(self.clone()), ) .await .map_err(DatabaseError::QueryError)?; Ok(()) } } #[derive(Debug, Clone, PartialEq, Eq)] pub enum UserIncidentType { Warning, Ban, } #[derive(Debug, Clone)] pub struct UserIncident { pub user_id: u64, pub incident_id: u64, pub report_id: Option, pub r#type: UserIncidentType, pub message: Option, pub reason: String, pub created_by_id: Option, pub created_at: UtcDateTime, pub expires_at: Option, pub duration: Option, } impl<'a> Object<'a> for UserIncident { type Row = ( i64, i64, Option, String, Option, String, Option, OffsetDateTime, Option, Option, ); fn from_row(row: Self::Row) -> Self { Self { user_id: i64_to_u64(row.0), incident_id: i64_to_u64(row.1), report_id: row.2.map(i64_to_u64), r#type: match row.3.as_str() { "warning" => UserIncidentType::Warning, "ban" => UserIncidentType::Ban, _ => unreachable!(), }, message: row.4, reason: row.5, created_by_id: row.6.map(i64_to_u64), created_at: row.7.to_utc(), expires_at: row.8.map(|i| i.to_utc()), duration: row.9, } } fn to_row(value: Self) -> Self::Row { ( u64_to_i64(value.user_id), u64_to_i64(value.incident_id), value.report_id.map(u64_to_i64), match value.r#type { UserIncidentType::Warning => "warning".into(), UserIncidentType::Ban => "ban".into(), }, value.message, value.reason, value.created_by_id.map(u64_to_i64), value.created_at.to_offset(UtcOffset::UTC), value.expires_at.map(|i| i.to_offset(UtcOffset::UTC)), value.duration, ) } } impl UserIncident { pub fn is_active(&self) -> bool { self.expires_at .map(|e| e > UtcDateTime::now()) .unwrap_or(true) && self.r#type == UserIncidentType::Ban } pub async fn get_by_user_and_incident_id( user_id: u64, incident_id: u64, session: &Session, ) -> Result, DatabaseError> { let user_id = u64_to_i64(user_id); let incident_id = u64_to_i64(incident_id); let result = session .query_unpaged( r#" SELECT user_id, incident_id, report_id, type, message, reason, created_by_id, created_at, expires_at, duration FROM payphone.user_incidents WHERE user_id = ? AND incident_id = ?; "#, (user_id, incident_id), ) .await? .into_rows_result()?; get_single_row_from_result(&result).await } pub async fn get_for_user(user_id: u64, session: &Session) -> Result, DatabaseError> { let user_id = u64_to_i64(user_id); let result = session .query_unpaged( r#" SELECT user_id, incident_id, report_id, type, message, reason, created_by_id, created_at, expires_at, duration FROM payphone.user_incidents WHERE user_id = ?; "#, (user_id,), ) .await? .into_rows_result()?; Ok(result .rows::<::Row>()? .collect::, _>>()? .into_iter() .map(Self::from_row) .collect::>()) } pub async fn insert(&self, session: &Session) -> Result<(), DatabaseError> { session .query_unpaged( r#" INSERT INTO payphone.user_incidents ( user_id, incident_id, report_id, type, message, reason, created_by_id, created_at, expires_at, duration ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?); "#, Self::to_row(self.clone()), ) .await .map_err(DatabaseError::QueryError)?; Ok(()) } pub async fn delete_by_user_and_incident_id( user_id: u64, incident_id: u64, session: &Session, ) -> Result<(), DatabaseError> { let user_id = u64_to_i64(user_id); let incident_id = u64_to_i64(incident_id); session .query_unpaged( r#" DELETE FROM payphone.user_incidents WHERE user_id = ? AND incident_id = ?; "#, (user_id, incident_id), ) .await .map_err(DatabaseError::QueryError)?; Ok(()) } pub async fn expire_by_user_and_incident_id( user_id: u64, incident_id: u64, session: &Session, ) -> Result<(), DatabaseError> { let Some(mut incident) = Self::get_by_user_and_incident_id(user_id, incident_id, session).await? else { return Ok(()); }; incident.expires_at = Some(UtcDateTime::now()); incident.insert(session).await } } #[derive(Debug, Clone)] pub struct Participant { pub call_id: u64, pub user_id: u64, pub channel_id: u64, pub user_name: String, pub user_avatar_url: String, pub user_mask: String, pub is_masked: bool, } impl<'a> Object<'a> for Participant { type Row = (i64, i64, i64, String, String, String, bool); fn from_row(row: Self::Row) -> Self { Self { call_id: i64_to_u64(row.0), user_id: i64_to_u64(row.1), channel_id: i64_to_u64(row.2), user_name: row.3, user_avatar_url: row.4, user_mask: row.5, is_masked: row.6, } } fn to_row(value: Self) -> Self::Row { ( u64_to_i64(value.call_id), u64_to_i64(value.user_id), u64_to_i64(value.channel_id), value.user_name, value.user_avatar_url, value.user_mask, value.is_masked, ) } } fn make_name_valid(mut name: String) -> String { if let Err(error) = webhook_username(&name) { if let ValidationErrorType::WebhookUsername { len, substring } = error.kind() { if let Some(len) = len { if *len < 2 { name = format!("{name}--"); } else { name = format!("{name:.80}"); } } else { name = "Invalid name".into(); } } } name } impl Participant { pub fn display_name(&self) -> String { if self.is_masked { get_mask_display_name(self.user_mask.clone()) } else { make_name_valid(self.user_name.clone()) } } pub fn avatar_url(&self) -> String { if self.is_masked { get_mask_avatar_url(self.user_mask.clone()).expect("Failed to get mask avatar URL") } else { self.user_avatar_url.clone() } } pub async fn get_for_call(call_id: u64, session: &Session) -> Result, DatabaseError> { let call_id = u64_to_i64(call_id); let rows = session .query_unpaged( r#" SELECT call_id, user_id, channel_id, user_name, user_avatar_url, user_mask, is_masked FROM payphone.participants WHERE call_id = ?; "#, (call_id,), ) .await .map_err(DatabaseError::QueryError)?; Ok(rows .into_rows_result()? .rows()? .collect::, _>>()? .into_iter() .map(Self::from_row) .collect::>()) } pub async fn get_for_user_and_channel_in_call( user_id: u64, channel_id: u64, call_id: u64, session: &Session, ) -> Result, DatabaseError> { let user_id = u64_to_i64(user_id); let channel_id = u64_to_i64(channel_id); let call_id = u64_to_i64(call_id); let result = session .query_unpaged( r#" SELECT call_id, user_id, channel_id, user_name, user_avatar_url, user_mask, is_masked FROM payphone.participants WHERE call_id = ? AND user_id = ? AND channel_id = ? LIMIT 1; "#, (call_id, user_id, channel_id), ) .await? .into_rows_result()?; get_single_row_from_result(&result).await } pub async fn get_or_create( user_id: u64, channel_id: u64, call_id: u64, display_name: String, display_avatar_url: String, is_masked: bool, session: &Session, ) -> Result { let participants = Self::get_for_call(call_id, session).await?; if let Some(participant) = participants .iter() .find(|p| p.user_id == user_id && p.channel_id == channel_id) { // User is already a participant in the call return Ok(participant.clone()); } let masks_in_use = participants .iter() .filter_map(|p| { if p.channel_id == channel_id { Some(p.user_mask.clone()) } else { None } }) .collect::>(); let (mask, _) = get_random_mask(&masks_in_use).ok_or(DatabaseError::Unknown( "No remaining masks available for the channel", ))?; let participant = Participant { call_id, user_id, channel_id, user_name: display_name, user_avatar_url: display_avatar_url, user_mask: mask, is_masked, }; participant.insert(session).await?; Ok(participant) } pub async fn insert(&self, session: &Session) -> Result<(), DatabaseError> { session .query_unpaged( r#" INSERT INTO payphone.participants ( call_id, user_id, channel_id, user_name, user_avatar_url, user_mask, is_masked ) VALUES (?, ?, ?, ?, ?, ?, ?) IF NOT EXISTS; "#, Self::to_row(self.clone()), ) .await .map_err(DatabaseError::QueryError)?; Ok(()) } pub async fn upsert(&self, session: &Session) -> Result<(), DatabaseError> { session .query_unpaged( r#" INSERT INTO payphone.participants ( call_id, user_id, channel_id, user_name, user_avatar_url, user_mask, is_masked ) VALUES (?, ?, ?, ?, ?, ?, ?); "#, Self::to_row(self.clone()), ) .await .map_err(DatabaseError::QueryError)?; Ok(()) } }