nyeki/payphone
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add support for masks

Browse Source
This commit is contained in:
nyeki
2025-08-24 02:02:25 -03:00
parent 2a2ffd8894
commit 325b4cbc16
8 changed files with 454 additions and 221 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/database/migrations/0005_call_messages_by_author_id.cql
+4
View File
@@ -0,0 +1,4 @@
CREATE MATERIALIZED VIEW IF NOT EXISTS payphone.messages_by_author_id AS
SELECT call_id, message_id, author_id FROM payphone.messages
WHERE call_id IS NOT NULL AND message_id IS NOT NULL AND author_id IS NOT NULL
PRIMARY KEY (call_id, author_id, message_id);
src/database/models.rs
+276 -111
View File
@@ -2,7 +2,10 @@ use scylla::{
DeserializeValue, SerializeValue,
client::session::Session,
deserialize::row::DeserializeRow,
errors::{ExecutionError, SingleRowError},
errors::{
DeserializationError, ExecutionError, FirstRowError, IntoRowsResultError, RowsError,
SingleRowError,
},
response::query_result::QueryRowsResult,
statement::{
Statement,
@@ -21,7 +24,7 @@ use twilight_model::{
use crate::{
constants::DEFAULT_PREFIX,
groups::calls::{DisplayProfile, masks::get_random_mask},
groups::calls::masks::{get_mask_avatar_url, get_mask_display_name, get_random_mask},
utils::{get_random_u64, i64_to_u64, u64_to_i64},
};
@@ -40,13 +43,22 @@ pub enum DatabaseOrDiscordError {
#[derive(Debug, Error)]
pub enum DatabaseError {
#[error("An error occurred when querying the database")]
QueryError(ExecutionError),
QueryError(#[from] ExecutionError),
#[error("The database responded but the result of the query weren't rows")]
ResultNotRows,
ResultNotRows(#[from] IntoRowsResultError),
#[error("The resulting row could not be deserialized into an object")]
DeserializationFailed,
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),
@@ -72,10 +84,8 @@ async fn get_single_row_from_result<'a, T: Object<'a>>(
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),
_ => Err(error.into()),
},
}
}
@@ -170,10 +180,8 @@ impl User {
"#,
(id,),
)
.await
.map_err(DatabaseError::QueryError)?
.into_rows_result()
.map_err(|_| DatabaseError::ResultNotRows)?;
.await?
.into_rows_result()?;
get_single_row_from_result(&result).await
}
@@ -301,10 +309,8 @@ impl User {
"#,
(),
)
.await
.map_err(DatabaseError::QueryError)?
.into_rows_result()
.map_err(|_| DatabaseError::ResultNotRows)?;
.await?
.into_rows_result()?;
Ok(result.rows_num())
}
@@ -364,10 +370,8 @@ impl UserByName {
"#,
(name,),
)
.await
.map_err(DatabaseError::QueryError)?
.into_rows_result()
.map_err(|_| DatabaseError::ResultNotRows)?;
.await?
.into_rows_result()?;
get_single_row_from_result(&result).await
}
@@ -423,20 +427,10 @@ impl Server {
"#,
(id,),
)
.await
.map_err(DatabaseError::QueryError)?
.into_rows_result()
.map_err(|_| DatabaseError::ResultNotRows)?;
.await?
.into_rows_result()?;
match result.single_row::<<Server as Object>::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(
@@ -643,10 +637,8 @@ impl Channel {
"#,
(id,),
)
.await
.map_err(DatabaseError::QueryError)?
.into_rows_result()
.map_err(|_| DatabaseError::ResultNotRows)?;
.await?
.into_rows_result()?;
get_single_row_from_result(&result).await
}
@@ -789,16 +781,12 @@ impl SearchingChannel {
"#,
(),
)
.await
.map_err(DatabaseError::QueryError)?
.into_rows_result()
.map_err(|_| DatabaseError::ResultNotRows)?;
.await?
.into_rows_result()?;
Ok(result
.rows::<<SearchingChannel as Object>::Row>()
.map_err(|_| DatabaseError::ResultNotRows)?
.collect::<Result<Vec<_>, _>>()
.map_err(|_| DatabaseError::DeserializationFailed)?
.rows::<<SearchingChannel as Object>::Row>()?
.collect::<Result<Vec<_>, _>>()?
.into_iter()
.map(Self::from_row)
.collect::<Vec<_>>())
@@ -865,8 +853,7 @@ impl Call {
)
.await
.map_err(DatabaseError::QueryError)?
.into_rows_result()
.map_err(|_| DatabaseError::ResultNotRows)?;
.into_rows_result()?;
get_single_row_from_result(&result).await
}
@@ -952,14 +939,11 @@ impl Call {
),
)
.await
.inspect_err(|e| log::error!("Failed to add channel to the call! {e:?}"))
.map_err(DatabaseError::QueryError)?
.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:?}"))
.map_err(|_| DatabaseError::ResultNotRows)?
.inspect_err(|e| log::error!("Failed to get result of the query! {e:?}"))?
.first_row::<(bool, Option<i64>)>()
.inspect_err(|e| log::error!("Failed to get first row of the result! {e:?}"))
.map_err(|_| DatabaseError::DeserializationFailed)?;
.inspect_err(|e| log::error!("Failed to get first row of the result! {e:?}"))?;
log::trace!("Is applied: {is_applied:?}");
@@ -1075,10 +1059,8 @@ impl Call {
"#,
(),
)
.await
.map_err(DatabaseError::QueryError)?
.into_rows_result()
.map_err(|_| DatabaseError::ResultNotRows)?;
.await?
.into_rows_result()?;
Ok(result.rows_num())
}
@@ -1115,7 +1097,7 @@ pub struct MessageBuilderError;
#[derive(Default)]
pub struct MessageBuilder {
message: Option<TwilightMessage>,
profile: Option<DisplayProfile>,
participant: Option<Participant>,
call_id: Option<u64>,
is_censored: bool,
propagation_message_id: Option<u64>,
@@ -1127,7 +1109,7 @@ impl MessageBuilder {
fn new() -> Self {
Self {
message: None,
profile: None,
participant: None,
call_id: None,
is_censored: false,
propagation_message_id: None,
@@ -1141,8 +1123,8 @@ impl MessageBuilder {
self
}
pub fn profile(mut self, profile: DisplayProfile) -> Self {
self.profile = Some(profile);
pub fn participant(mut self, participant: Participant) -> Self {
self.participant = Some(participant);
self
}
@@ -1176,16 +1158,16 @@ impl MessageBuilder {
pub fn build(self) -> Result<Message, MessageBuilderError> {
let message = self.message.ok_or(MessageBuilderError)?;
let profile = self.profile.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(profile.id),
author_name: profile.name.clone(),
author_avatar_url: profile.avatar_url.clone(),
author_mask: profile.mask,
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
@@ -1206,7 +1188,7 @@ impl MessageBuilder {
Ok(Message {
call_id,
message_id: message.id.get(),
author_id: profile.id,
author_id: participant.user_id,
propagation_message_id: self.propagation_message_id,
data,
})
@@ -1276,10 +1258,8 @@ impl Message {
"#,
(call_id, message_id_i64),
)
.await
.map_err(DatabaseError::QueryError)?
.into_rows_result()
.map_err(|_| DatabaseError::ResultNotRows)?;
.await?
.into_rows_result()?;
if result.rows_num() == 0 {
Self::get_by_propagation_message_id(message_id, session).await
@@ -1317,10 +1297,46 @@ impl Message {
u64_to_i64(message_by_propagation_id.message_id),
),
)
.await
.map_err(DatabaseError::QueryError)?
.into_rows_result()
.map_err(|_| DatabaseError::ResultNotRows)?;
.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<Option<Self>, 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;
}
@@ -1356,10 +1372,8 @@ impl Message {
"#,
(),
)
.await
.map_err(DatabaseError::QueryError)?
.into_rows_result()
.map_err(|_| DatabaseError::ResultNotRows)?;
.await?
.into_rows_result()?;
Ok(result.rows_num())
}
@@ -1413,10 +1427,67 @@ impl MessageByPropagationId {
"#,
(propagation_message_id,),
)
.await
.map_err(DatabaseError::QueryError)?
.into_rows_result()
.map_err(|_| DatabaseError::ResultNotRows)?;
.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<Option<Self>, 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
}
@@ -1637,10 +1708,8 @@ impl UserIncident {
"#,
(user_id, incident_id),
)
.await
.map_err(DatabaseError::QueryError)?
.into_rows_result()
.map_err(|_| DatabaseError::ResultNotRows)?;
.await?
.into_rows_result()?;
get_single_row_from_result(&result).await
}
@@ -1669,16 +1738,12 @@ impl UserIncident {
"#,
(user_id,),
)
.await
.map_err(DatabaseError::QueryError)?
.into_rows_result()
.map_err(|_| DatabaseError::ResultNotRows)?;
.await?
.into_rows_result()?;
Ok(result
.rows::<<Self as Object>::Row>()
.map_err(|_| DatabaseError::ResultNotRows)?
.collect::<Result<Vec<_>, _>>()
.map_err(|_| DatabaseError::DeserializationFailed)?
.rows::<<Self as Object>::Row>()?
.collect::<Result<Vec<_>, _>>()?
.into_iter()
.map(Self::from_row)
.collect::<Vec<_>>())
@@ -1789,14 +1854,48 @@ impl<'a> Object<'a> for Participant {
}
}
fn make_name_valid(name: String) -> String {
if name.len() == 1 {
format!("{name}\u{2800}")
} else if name.len() > 80 {
format!("{name:.80}")
} else {
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<Vec<Self>, DatabaseError> {
let call_id = u64_to_i64(call_id);
let rows = session
.query_unpaged(
r#"
SELECT * FROM payphone.participants
SELECT
call_id,
user_id,
channel_id,
user_name,
user_avatar_url,
user_mask,
is_masked
FROM payphone.participants
WHERE call_id = ?;
"#,
(call_id,),
@@ -1805,27 +1904,69 @@ impl Participant {
.map_err(DatabaseError::QueryError)?;
Ok(rows
.into_rows_result()
.map_err(|_| DatabaseError::ResultNotRows)?
.rows()
.map_err(|_| DatabaseError::ResultNotRows)?
.collect::<Result<Vec<_>, _>>()
.map_err(|_| DatabaseError::DeserializationFailed)?
.into_rows_result()?
.rows()?
.collect::<Result<Vec<_>, _>>()?
.into_iter()
.map(Self::from_row)
.collect::<Vec<_>>())
}
pub async fn create(
pub async fn get_for_user_and_channel_in_call(
user_id: u64,
channel_id: u64,
call_id: u64,
profile: DisplayProfile,
session: &Session,
) -> Result<Option<Self>, 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<(), DatabaseError> {
let masks_in_use = Self::get_for_call(call_id, session)
.await?
) -> Result<Self, DatabaseError> {
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 {
@@ -1844,13 +1985,15 @@ impl Participant {
call_id,
user_id,
channel_id,
user_name: profile.name,
user_avatar_url: profile.avatar_url,
user_name: display_name,
user_avatar_url: display_avatar_url,
user_mask: mask,
is_masked,
};
participant.insert(session).await
participant.insert(session).await?;
Ok(participant)
}
pub async fn insert(&self, session: &Session) -> Result<(), DatabaseError> {
@@ -1865,7 +2008,29 @@ impl Participant {
user_avatar_url,
user_mask,
is_masked
) IF NOT EXISTS VALUES (?, ?, ?, ?, ?, ?, ?);
) 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()),
)