Add support for replying on call

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<payphone.message>,
 
     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,
 

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<Option<T>, 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<Option<T>, DatabaseError> {
+    match result.single_row::<T::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::<<Channel as Object>::Row>() {
+        get_single_row_from_result(&result).await
-            Ok(row) => Ok(Some(Self::from_row(row))),
-            Err(error) => match error {
-                SingleRowError::DeserializationFailed(_) | SingleRowError::TypeCheckFailed(_) => {
-                    Err(DatabaseError::DeserializationFailed)
-                }
-                SingleRowError::UnexpectedRowCount(_) => Ok(None),
-            },
-        }
     }
 
     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::<<Call as Object>::Row>() {
+        get_single_row_from_result(&result).await
-            Ok(row) => Ok(Some(Self::from_row(row))),
-            Err(error) => match error {
-                SingleRowError::DeserializationFailed(_) | SingleRowError::TypeCheckFailed(_) => {
-                    Err(DatabaseError::DeserializationFailed)
-                }
-                SingleRowError::UnexpectedRowCount(_) => Ok(None),
-            },
-        }
     }
 
     /// 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 {
+    // Argument is made optional here because it happens to be more convenient for where this is
-        self.replying_to_id = Some(replying_to_id);
+    // used.
+    pub fn replying_to_id(mut self, replying_to_id: Option<u64>) -> 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<u64>,
     pub data: MessageData,
 }
 
 impl<'a> Object<'a> for Message {
-    type Row = (i64, i64, i64, MessageData);
+    type Row = (i64, i64, i64, Option<i64>, 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<Option<Self>, 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::<<Message as Object>::Row>() {
+        if result.rows_num() == 0 {
-            Ok(row) => Ok(Some(Self::from_row(row))),
+            let message_by_propagation_id =
-            Err(error) => match error {
+                MessageByPropagationId::get_by_propagation_message_id(message_id, session).await?;
-                SingleRowError::DeserializationFailed(_) | SingleRowError::TypeCheckFailed(_) => {
+
-                    Err(DatabaseError::DeserializationFailed)
+            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;
             }
-                SingleRowError::UnexpectedRowCount(_) => Ok(None),
-            },
         }
+
+        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<Option<Self>, 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,

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<State> 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<State> 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<State> 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<State> 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()