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::{ channel::Message as TwilightMessage, guild::{Guild as TwilightGuild, GuildInfo as TwilightGuildInfo}, id::Id, user::User as TwilightUser, }; use crate::{constants::DEFAULT_PREFIX, groups::calls::DisplayProfile, 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 { #[allow(dead_code)] 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)] 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 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 ) 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(()) } } #[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_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, profile: Option, call_id: Option, } impl MessageBuilder { fn new() -> Self { Self { message: None, profile: None, call_id: None, } } pub fn message(mut self, message: TwilightMessage) -> Self { self.message = Some(message); self } pub fn profile(mut self, profile: DisplayProfile) -> Self { self.profile = Some(profile); self } pub fn call_id(mut self, call_id: u64) -> Self { self.call_id = Some(call_id); self } pub fn build(self) -> Result { let message = self.message.ok_or(MessageBuilderError)?; let profile = self.profile.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, 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: None, propagation_channel_id: None, replying_to_id: None, is_deleted: false, is_edited: message.edited_timestamp.is_some(), is_censored: false, }; Ok(Message { call_id, message_id: message.id.get(), author_id: profile.id, data, }) } } #[derive(Debug, Clone)] pub struct Message { pub call_id: u64, pub message_id: u64, pub author_id: u64, pub data: MessageData, } impl<'a> Object<'a> for Message { type Row = (i64, i64, 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, } } 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.data, ) } } impl Message { pub fn build() -> MessageBuilder { MessageBuilder::new() } pub async fn insert(&self, session: &Session) -> Result<(), DatabaseError> { session .query_unpaged( r#" INSERT INTO payphone.messages ( call_id, message_id, author_id, data ) VALUES (?, ?, ?, ?); "#, Self::to_row(self.clone()), ) .await .map_err(DatabaseError::QueryError)?; Ok(()) } }