use scylla::{ DeserializeValue, SerializeValue, client::session::Session, deserialize::row::DeserializeRow, errors::{ExecutionError, SingleRowError}, statement::{ Statement, batch::{Batch, BatchType}, }, }; use thiserror::Error; use time::{OffsetDateTime, UtcDateTime}; use twilight_http::Client; use twilight_model::{ guild::{Guild as TwilightGuild, GuildInfo as TwilightGuildInfo}, id::Id, user::User as TwilightUser, }; use crate::{constants::DEFAULT_PREFIX, utils::get_random_u64}; #[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(ExecutionError), #[error("The database responded but the result of the query weren't rows")] ResultNotRows, #[error("The resulting row could not be deserialized into an object")] DeserializationFailed, } /// Copies the bytes from [`i64`] to [`u64`]. /// /// WARNING: This does NOT keep the same "represented" value. This only keeps the bytes intact, /// e.g. 0b0101 will still be 0b0101, but the value represented will change since the type changes. /// /// Arguments: /// * `i` - The [`i64`] to convert. /// /// Returns: /// [`u64`] - The resulting [`u64`]. fn i64_to_u64(i: i64) -> u64 { u64::from_ne_bytes(i.to_ne_bytes()) } /// Copies the bytes from [`u64`] to [`i64`]. /// /// WARNING: This does NOT keep the same "represented" value. This only keeps the bytes intact, /// e.g. 0b0101 will still be 0b0101, but the value represented will change since the type changes. /// /// Arguments: /// * `u` - The [`u64`] to convert. /// /// Returns: /// [`i64`] - The resulting [`i64`]. fn u64_to_i64(u: u64) -> i64 { i64::from_ne_bytes(u.to_ne_bytes()) } trait Object<'a> { type Row: DeserializeRow<'a, 'a>; fn from_row(row: Self::Row) -> Self; fn to_row(value: Self) -> Self::Row; } #[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, } impl<'a> Object<'a> for User { type Row = ( i64, String, Option, bool, bool, OffsetDateTime, OffsetDateTime, 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()), } } 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()), ) } } 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 FROM payphone.users WHERE user_id = ? LIMIT 1; "#, (id,), ) .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), }, } } } 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, } } } #[derive(Debug, Clone, SerializeValue, DeserializeValue)] pub struct Message { pub id: i64, pub author_id: i64, pub author_name: String, pub author_avatar_url: String, pub author_mask: Option, pub content: String, pub attachments: Option>, pub sticker: Option, pub propagation_message_id: Option, pub propagation_channel_id: Option, pub replying_to_id: Option, pub is_deleted: bool, pub is_edited: bool, pub is_censored: bool, } impl<'a> Object<'a> for Message { type Row = ( i64, i64, String, String, Option, String, Option>, Option, Option, Option, Option, bool, bool, bool, ); fn from_row(row: Self::Row) -> Self { Self { id: row.0, author_id: row.1, author_name: row.2, author_avatar_url: row.3, author_mask: row.4, content: row.5, attachments: row.6, sticker: row.7, propagation_message_id: row.8, propagation_channel_id: row.9, replying_to_id: row.10, is_deleted: row.11, is_edited: row.12, is_censored: row.13, } } fn to_row(value: Self) -> Self::Row { ( value.id, value.author_id, value.author_name, value.author_avatar_url, value.author_mask, value.content, value.attachments, value.sticker, value.propagation_message_id, value.propagation_channel_id, value.replying_to_id, value.is_deleted, value.is_edited, value.is_censored, ) } } pub struct MessageRow { pub call_id: u64, pub message_id: u64, pub data: Message, } impl MessageRow { 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 = u64_to_i64(message_id); let result = session .query_unpaged( r#" SELECT call_id, message_id, data FROM payphone.messages WHERE call_id = ? AND message_id = ? LIMIT 1; "#, (call_id, message_id), ) .await .map_err(DatabaseError::QueryError)? .into_rows_result() .map_err(|_| DatabaseError::ResultNotRows)?; match result.single_row::<::Row>() { Ok(row) => Ok(Some(::from_row(row))), Err(error) => match error { SingleRowError::DeserializationFailed(_) | SingleRowError::TypeCheckFailed(_) => { Err(DatabaseError::DeserializationFailed) } SingleRowError::UnexpectedRowCount(_) => Ok(None), }, } } } impl<'a> Object<'a> for Report { type Row = ( i64, // report_id Message, Option, OffsetDateTime, Option, Vec, Option, Option, ); fn from_row(row: Self::Row) -> Self { Self { report_id: i64_to_u64(row.0), message: row.1, replying_to: row.2, created_at: row.3, reviewed_at: row.4, created_by_ids: row.5.iter().map(|&id| i64_to_u64(id)).collect(), reviewed_by_id: row.6.map(i64_to_u64), resolution: row.7, } } fn to_row(value: Self) -> Self::Row { ( u64_to_i64(value.report_id), value.message, value.replying_to, value.created_at, value.reviewed_at, value .created_by_ids .iter() .map(|&id| u64_to_i64(id)) .collect(), value.reviewed_by_id.map(u64_to_i64), value.resolution, ) } } // --- BEGIN: Payphone Message and Report Models --- #[derive(Debug, Clone)] pub struct Report { pub report_id: u64, pub message: Message, pub replying_to: Option, pub created_at: OffsetDateTime, pub reviewed_at: Option, pub created_by_ids: Vec, pub reviewed_by_id: Option, pub resolution: Option, } impl<'a> Object<'a> for MessageRow { type Row = (i64, i64, >::Row); fn from_row(row: Self::Row) -> Self { Self { call_id: i64_to_u64(row.0), message_id: i64_to_u64(row.1), data: ::from_row(row.2), } } fn to_row(value: Self) -> Self::Row { ( u64_to_i64(value.call_id), u64_to_i64(value.message_id), ::to_row(value.data), ) } } impl Report { pub async fn insert(&self, session: &Session) -> Result<(), DatabaseError> { let row = Self::to_row(self.clone()); session .query_unpaged( r#" INSERT INTO payphone.reports ( report_id, message, replying_to, created_at, reviewed_at, created_by_ids, reviewed_by_id, resolution ) VALUES (?, ?, ?, ?, ?, ?, ?, ?); "#, row, ) .await .map_err(DatabaseError::QueryError)?; Ok(()) } } #[derive(Debug, Clone)] pub struct Server { pub id: u64, pub name: String, pub prefix: String, } impl<'a> Object<'a> for Server { type Row = (i64, String, String); fn from_row(row: Self::Row) -> Self { Self { id: i64_to_u64(row.0), name: row.1, prefix: row.2, } } fn to_row(value: Self) -> Self::Row { (u64_to_i64(value.id), value.name, value.prefix) } } 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 FROM payphone.servers WHERE server_id = ? LIMIT 1; "#, (id,), ) .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), }, } } pub async fn insert(&self, session: &Session) -> Result<(), DatabaseError> { session .query_unpaged( r#" INSERT INTO payphone.servers ( server_id, name, prefix ) 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 ) 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(), } } } impl From for Server { fn from(value: TwilightGuildInfo) -> Self { Self { id: value.id.get(), name: value.name, prefix: DEFAULT_PREFIX.into(), } } } #[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, ) } } 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 .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), }, } } 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 = 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 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(()) } } #[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 .map_err(DatabaseError::QueryError)? .into_rows_result() .map_err(|_| DatabaseError::ResultNotRows)?; Ok(result .rows::<::Row>() .map_err(|_| DatabaseError::ResultNotRows)? .collect::, _>>() .map_err(|_| DatabaseError::DeserializationFailed)? .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() .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), }, } } /// 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:?}")) .map_err(DatabaseError::QueryError)? .into_rows_result() .inspect_err(|e| log::error!("Failed to get result of the query! {e:?}")) .map_err(|_| DatabaseError::ResultNotRows)? .first_row::<(bool, Option)>() .inspect_err(|e| log::error!("Failed to get first row of the result! {e:?}")) .map_err(|_| DatabaseError::DeserializationFailed)?; 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(()) } }