diff --git a/src/database/migrations/0001_init.cql b/src/database/migrations/0001_init.cql index c1fcba9..ee29466 100644 --- a/src/database/migrations/0001_init.cql +++ b/src/database/migrations/0001_init.cql @@ -88,12 +88,18 @@ CREATE TABLE IF NOT EXISTS payphone.messages ( call_id BIGINT, message_id BIGINT, author_id BIGINT, + propagation_message_id BIGINT, data FROZEN, PRIMARY KEY (call_id, message_id) ) WITH default_time_to_live = 2592000; -- 30 days +CREATE MATERIALIZED VIEW IF NOT EXISTS payphone.messages_by_propagation_id AS + SELECT call_id, message_id, propagation_message_id FROM payphone.messages + WHERE call_id IS NOT NULL AND message_id IS NOT NULL AND propagation_message_id IS NOT NULL + PRIMARY KEY (call_id, propagation_message_id, message_id); + CREATE TABLE IF NOT EXISTS payphone.servers ( server_id BIGINT PRIMARY KEY, diff --git a/src/database/models.rs b/src/database/models.rs index 2bf979a..cc04f20 100644 --- a/src/database/models.rs +++ b/src/database/models.rs @@ -3,6 +3,7 @@ use scylla::{ client::session::Session, deserialize::row::DeserializeRow, errors::{ExecutionError, SingleRowError}, + response::query_result::QueryRowsResult, statement::{ Statement, batch::{Batch, BatchType}, @@ -55,6 +56,27 @@ trait Object<'a> { 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::DeserializationFailed(_) | SingleRowError::TypeCheckFailed(_) => { + Err(DatabaseError::DeserializationFailed) + } + SingleRowError::UnexpectedRowCount(_) => Ok(None), + }, + } +} + #[derive(Debug, Clone)] pub struct User { pub id: u64, @@ -484,15 +506,7 @@ impl Channel { .into_rows_result() .map_err(|_| DatabaseError::ResultNotRows)?; - match result.single_row::<::Row>() { - Ok(row) => Ok(Some(Self::from_row(row))), - Err(error) => match error { - SingleRowError::DeserializationFailed(_) | SingleRowError::TypeCheckFailed(_) => { - Err(DatabaseError::DeserializationFailed) - } - SingleRowError::UnexpectedRowCount(_) => Ok(None), - }, - } + get_single_row_from_result(&result).await } pub async fn get_or_create_by_id( @@ -705,15 +719,7 @@ impl Call { .into_rows_result() .map_err(|_| DatabaseError::ResultNotRows)?; - match result.single_row::<::Row>() { - Ok(row) => Ok(Some(Self::from_row(row))), - Err(error) => match error { - SingleRowError::DeserializationFailed(_) | SingleRowError::TypeCheckFailed(_) => { - Err(DatabaseError::DeserializationFailed) - } - SingleRowError::UnexpectedRowCount(_) => Ok(None), - }, - } + get_single_row_from_result(&result).await } /// Returns the opposite party's channel ID. @@ -992,8 +998,10 @@ impl MessageBuilder { self } - pub fn replying_to_id(mut self, replying_to_id: u64) -> Self { - self.replying_to_id = Some(replying_to_id); + // Argument is made optional here because it happens to be more convenient for where this is + // used. + pub fn replying_to_id(mut self, replying_to_id: Option) -> Self { + self.replying_to_id = replying_to_id; self } @@ -1030,6 +1038,7 @@ impl MessageBuilder { call_id, message_id: message.id.get(), author_id: profile.id, + propagation_message_id: self.propagation_message_id, data, }) } @@ -1040,18 +1049,20 @@ 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, MessageData); + 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), - data: row.3, + propagation_message_id: row.3.map(i64_to_u64), + data: row.4, } } @@ -1060,6 +1071,7 @@ impl<'a> Object<'a> for Message { 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, ) } @@ -1076,7 +1088,7 @@ impl Message { session: &Session, ) -> Result, DatabaseError> { let call_id = u64_to_i64(call_id); - let message_id = u64_to_i64(message_id); + let message_id_i64 = u64_to_i64(message_id); let result = session .query_unpaged( @@ -1085,6 +1097,7 @@ impl Message { call_id, message_id, author_id, + propagation_message_id, data FROM payphone.messages @@ -1092,22 +1105,45 @@ impl Message { call_id = ? AND message_id = ? LIMIT 1; "#, - (call_id, message_id), + (call_id, message_id_i64), ) .await .map_err(DatabaseError::QueryError)? .into_rows_result() .map_err(|_| DatabaseError::ResultNotRows)?; - match result.single_row::<::Row>() { - Ok(row) => Ok(Some(Self::from_row(row))), - Err(error) => match error { - SingleRowError::DeserializationFailed(_) | SingleRowError::TypeCheckFailed(_) => { - Err(DatabaseError::DeserializationFailed) - } - SingleRowError::UnexpectedRowCount(_) => Ok(None), - }, + if result.rows_num() == 0 { + let message_by_propagation_id = + MessageByPropagationId::get_by_propagation_message_id(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; + "#, + (call_id, u64_to_i64(message_by_propagation_id.message_id)), + ) + .await + .map_err(DatabaseError::QueryError)? + .into_rows_result() + .map_err(|_| DatabaseError::ResultNotRows)?; + + return get_single_row_from_result(&result).await; + } } + + get_single_row_from_result(&result).await } pub async fn insert(&self, session: &Session) -> Result<(), DatabaseError> { @@ -1118,8 +1154,9 @@ impl Message { call_id, message_id, author_id, + propagation_message_id, data - ) VALUES (?, ?, ?, ?); + ) VALUES (?, ?, ?, ?, ?); "#, Self::to_row(self.clone()), ) @@ -1130,6 +1167,63 @@ impl Message { } } +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 + .map_err(DatabaseError::QueryError)? + .into_rows_result() + .map_err(|_| DatabaseError::ResultNotRows)?; + + get_single_row_from_result(&result).await + } +} + #[derive(Debug, Clone, PartialEq, Eq)] pub enum UserIncidentType { Warning, diff --git a/src/groups/calls/events.rs b/src/groups/calls/events.rs index 9c41255..4d4d941 100644 --- a/src/groups/calls/events.rs +++ b/src/groups/calls/events.rs @@ -6,8 +6,10 @@ use twilight_model::{ gateway::payload::incoming::{MessageCreate, MessageDelete, MessageUpdate}, id::Id, }; +use twilight_util::builder::embed::{EmbedAuthorBuilder, EmbedBuilder, ImageSource}; use crate::{ + constants::COLOR_WHITE, database::models::{Call, Channel, Message, User, UserIncident}, discord::sanitizing::SanitizeDiscordMarkdown, framework::{ @@ -23,7 +25,7 @@ use crate::{ moderation::{Action, Rule, escape_message}, }, state::State, - utils::i64_to_u64, + utils::{ellipse, i64_to_u64}, }; async fn penalize_user( @@ -177,6 +179,22 @@ impl OnMessageHandler for CallMessageHandler { let message: Message; + let replying_to = if let Some(reference) = &event.reference { + if let Some(replying_to_message_id) = reference.message_id { + Message::get_by_call_and_message_id( + call.id, + replying_to_message_id.get(), + &rt.state.db, + ) + .await? + } else { + None + } + } else { + None + }; + let replying_to_id = replying_to.as_ref().map(|i| i.message_id); + if rules .iter() .any(|rule| rule.action() == &Action::BlockAndPenalty) @@ -201,11 +219,28 @@ impl OnMessageHandler for CallMessageHandler { .message(event.0.clone()) .profile(profile.clone()) .is_censored(!rules.is_empty()) + .replying_to_id(replying_to_id) .build() .expect( "The message builder was missing either the call ID, the message, or the profile", ); } else { + let mut embeds = vec![]; + + if let Some(replying_to) = replying_to { + embeds.push( + EmbedBuilder::new() + .author( + EmbedAuthorBuilder::new(replying_to.data.author_name).icon_url( + ImageSource::url(replying_to.data.author_avatar_url).unwrap(), + ), + ) + .color(COLOR_WHITE) + .description(ellipse(replying_to.data.content, 100)) + .build(), + ); + } + let message_sent = rt .http .execute_webhook(Id::new(webhook_id), &webhook_token) @@ -213,6 +248,7 @@ impl OnMessageHandler for CallMessageHandler { .username(&profile.name) .avatar_url(&profile.avatar_url) .content(&content) + .embeds(&embeds) .wait() .await .inspect_err(|e| { @@ -230,6 +266,7 @@ impl OnMessageHandler for CallMessageHandler { .message(event.0.clone()) .profile(profile.clone()) .is_censored(!rules.is_empty()) + .replying_to_id(replying_to_id) .propagation_message_id(message_sent.id.get()) .propagation_channel_id(opposite_channel.id) .build()