Finish refactoring
This commit is contained in:
@@ -2,7 +2,9 @@ CREATE KEYSPACE IF NOT EXISTS payphone
|
||||
WITH replication = {
|
||||
'class': 'NetworkTopologyStrategy',
|
||||
'replication_factor': 3
|
||||
};
|
||||
} AND TABLETS = {
|
||||
'enabled': false
|
||||
};
|
||||
|
||||
CREATE TABLE IF NOT EXISTS payphone.users (
|
||||
user_id BIGINT PRIMARY KEY,
|
||||
@@ -15,13 +17,15 @@ CREATE TABLE IF NOT EXISTS payphone.users (
|
||||
|
||||
joined_at TIMESTAMP,
|
||||
last_seen_at TIMESTAMP,
|
||||
is_supporter_since TIMESTAMP,
|
||||
is_supporter_until TIMESTAMP
|
||||
supporter_since TIMESTAMP,
|
||||
supporter_until TIMESTAMP
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS payphone.calls (
|
||||
call_id BIGINT PRIMARY KEY,
|
||||
|
||||
channel_ids LIST<BIGINT>,
|
||||
|
||||
started_at TIMESTAMP,
|
||||
finished_at TIMESTAMP,
|
||||
);
|
||||
@@ -30,22 +34,30 @@ CREATE TABLE IF NOT EXISTS payphone.channels (
|
||||
channel_id BIGINT PRIMARY KEY,
|
||||
server_id BIGINT,
|
||||
|
||||
name TEXT,
|
||||
|
||||
call_id BIGINT,
|
||||
is_searching BOOLEAN,
|
||||
|
||||
recently_called_channel_ids LIST<BIGINT>
|
||||
recently_called_server_ids LIST<BIGINT>,
|
||||
|
||||
webhook_id BIGINT,
|
||||
webhook_token TEXT
|
||||
);
|
||||
|
||||
CREATE MATERIALIZED VIEW IF NOT EXISTS payphone.searching_channels AS
|
||||
SELECT channel_id, server_id, is_searching, recently_called_server_ids FROM payphone.channels
|
||||
WHERE channel_id IS NOT NULL AND is_searching = true
|
||||
PRIMARY KEY (is_searching, channel_id);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS payphone.participants (
|
||||
call_id BIGINT,
|
||||
user_id BIGINT,
|
||||
channel_id BIGINT,
|
||||
|
||||
user_name TEXT,
|
||||
user_avatar_url TEXT,
|
||||
user_mask TEXT,
|
||||
|
||||
PRIMARY KEY (call_id, user_id)
|
||||
PRIMARY KEY (call_id, user_id, channel_id)
|
||||
);
|
||||
|
||||
-- TTL 30 days.
|
||||
@@ -62,6 +74,9 @@ CREATE TABLE IF NOT EXISTS payphone.messages (
|
||||
attachments LIST<TEXT>,
|
||||
sticker BIGINT,
|
||||
|
||||
propagation_message_id BIGINT,
|
||||
propagation_channel_id BIGINT,
|
||||
|
||||
replying_to_id BIGINT,
|
||||
|
||||
is_deleted BOOLEAN,
|
||||
|
||||
+61
-14
@@ -1,9 +1,12 @@
|
||||
pub mod models;
|
||||
|
||||
use include_dir::{Dir, include_dir};
|
||||
use scylla::{
|
||||
client::{session::Session, session_builder::SessionBuilder},
|
||||
errors::ExecutionError,
|
||||
};
|
||||
use thiserror::Error;
|
||||
use time::OffsetDateTime;
|
||||
|
||||
#[derive(Debug, Error)]
|
||||
pub enum DatabaseError {
|
||||
@@ -44,22 +47,66 @@ pub async fn init(uri: impl ToString) -> Result<Session, DatabaseError> {
|
||||
.into(),
|
||||
)
|
||||
.await
|
||||
.inspect(|_| log::info!("The migrations keyspace was successfully set up"))
|
||||
.inspect_err(|e| log::error!("Could not set up migrations keyspace and table! {e:?}"))
|
||||
.map_err(|_| DatabaseError::MigrationFailed)?;
|
||||
|
||||
for file in MIGRATIONS.files() {
|
||||
execute_file(
|
||||
&session,
|
||||
file.contents_utf8()
|
||||
.ok_or(DatabaseError::MigrationFileInvalid(
|
||||
file.path().to_string_lossy().to_string(),
|
||||
))?
|
||||
.to_string(),
|
||||
)
|
||||
.await
|
||||
.inspect_err(|e| log::error!("Could not migrate {}! {e:?}", file.path().to_str().unwrap()))
|
||||
.map_err(|_| DatabaseError::MigrationFailed)?;
|
||||
let mut sorted_migration_files = MIGRATIONS.files().collect::<Vec<_>>();
|
||||
sorted_migration_files.sort_by(|a, b| {
|
||||
a.path()
|
||||
.to_string_lossy()
|
||||
.to_string()
|
||||
.cmp(&b.path().to_string_lossy().to_string())
|
||||
});
|
||||
|
||||
for file in sorted_migration_files {
|
||||
let migration_name = file.path().to_string_lossy().to_string();
|
||||
|
||||
let row = session
|
||||
.query_unpaged(
|
||||
"SELECT name FROM migrations.applied WHERE name = ? LIMIT 1;",
|
||||
(&migration_name,),
|
||||
)
|
||||
.await
|
||||
.inspect_err(|e| {
|
||||
log::error!(
|
||||
"Could not check whether the migration \"{migration_name}\" was applied! {e:?}"
|
||||
)
|
||||
})
|
||||
.map_err(|_| DatabaseError::MigrationFailed)?
|
||||
.into_rows_result()
|
||||
.inspect_err(|e| log::error!("The applied migrations' database result were not rows! {e:?}"))
|
||||
.map_err(|_| DatabaseError::MigrationFailed)?
|
||||
.maybe_first_row::<(String,)>()
|
||||
.inspect_err(|e| log::error!("The applied migrations' row returned by the database could not be parsed! {e:?}"))
|
||||
.map_err(|_| DatabaseError::MigrationFailed)?;
|
||||
|
||||
if row.is_none() {
|
||||
execute_file(
|
||||
&session,
|
||||
file.contents_utf8()
|
||||
.ok_or(DatabaseError::MigrationFileInvalid(migration_name.clone()))?
|
||||
.to_string(),
|
||||
)
|
||||
.await
|
||||
.inspect(|_| log::debug!("Migration {migration_name} applied!"))
|
||||
.inspect_err(|e| {
|
||||
log::error!("Could not migrate {}! {e:?}", file.path().to_str().unwrap())
|
||||
})
|
||||
.map_err(|_| DatabaseError::MigrationFailed)?;
|
||||
|
||||
session
|
||||
.query_unpaged(
|
||||
"INSERT INTO migrations.applied (name, applied_at) VALUES (?, ?);",
|
||||
(migration_name.clone(), OffsetDateTime::now_utc()),
|
||||
)
|
||||
.await
|
||||
.inspect_err(|e| {
|
||||
log::error!("Could not mark migration {migration_name} as applied in DB! {e:?}")
|
||||
})
|
||||
.map_err(|_| DatabaseError::MigrationFailed)?;
|
||||
} else {
|
||||
log::trace!("Skipped migration {migration_name} because it was already applied.");
|
||||
}
|
||||
}
|
||||
|
||||
Ok(session)
|
||||
@@ -85,7 +132,7 @@ async fn execute_file(session: &Session, file: String) -> Result<(), FileExecuti
|
||||
.await
|
||||
.map_err(|error| FileExecutionError {
|
||||
error,
|
||||
statement_index,
|
||||
statement_index: statement_index + 1,
|
||||
})?;
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,828 @@
|
||||
use scylla::{
|
||||
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<String>,
|
||||
pub is_masked: bool,
|
||||
pub is_moderator: bool,
|
||||
pub joined_at: UtcDateTime,
|
||||
pub last_seen_at: UtcDateTime,
|
||||
pub supporter_since: Option<UtcDateTime>,
|
||||
pub supporter_until: Option<UtcDateTime>,
|
||||
}
|
||||
|
||||
impl<'a> Object<'a> for User {
|
||||
type Row = (
|
||||
i64,
|
||||
String,
|
||||
Option<String>,
|
||||
bool,
|
||||
bool,
|
||||
OffsetDateTime,
|
||||
OffsetDateTime,
|
||||
Option<OffsetDateTime>,
|
||||
Option<OffsetDateTime>,
|
||||
);
|
||||
|
||||
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<Option<Self>, 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::<<User 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),
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl From<TwilightUser> 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<Option<Self>, 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::<<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),
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
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<TwilightGuild> for Server {
|
||||
fn from(value: TwilightGuild) -> Self {
|
||||
Self {
|
||||
id: value.id.get(),
|
||||
name: value.name,
|
||||
prefix: DEFAULT_PREFIX.into(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl From<TwilightGuildInfo> 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<u64>,
|
||||
pub is_searching: bool,
|
||||
pub recently_called_server_ids: Vec<u64>,
|
||||
pub webhook_id: Option<u64>,
|
||||
pub webhook_token: Option<String>,
|
||||
}
|
||||
|
||||
impl<'a> Object<'a> for Channel {
|
||||
type Row = (
|
||||
i64,
|
||||
i64,
|
||||
Option<i64>,
|
||||
bool,
|
||||
Vec<i64>,
|
||||
Option<i64>,
|
||||
Option<String>,
|
||||
);
|
||||
|
||||
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<Option<Self>, 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::<<Channel 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),
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn get_or_create_by_id(
|
||||
id: u64,
|
||||
session: &Session,
|
||||
http: &Client,
|
||||
) -> Result<Self, DatabaseOrDiscordError> {
|
||||
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<u64>,
|
||||
}
|
||||
|
||||
impl<'a> Object<'a> for SearchingChannel {
|
||||
type Row = (i64, i64, Vec<i64>);
|
||||
|
||||
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<Vec<Self>, 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::<<SearchingChannel as Object>::Row>()
|
||||
.map_err(|_| DatabaseError::ResultNotRows)?
|
||||
.collect::<Result<Vec<_>, _>>()
|
||||
.map_err(|_| DatabaseError::DeserializationFailed)?
|
||||
.into_iter()
|
||||
.map(Self::from_row)
|
||||
.collect::<Vec<_>>())
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct Call {
|
||||
pub id: u64,
|
||||
pub channel_ids: Vec<u64>,
|
||||
pub started_at: UtcDateTime,
|
||||
pub finished_at: Option<UtcDateTime>,
|
||||
}
|
||||
|
||||
impl<'a> Object<'a> for Call {
|
||||
type Row = (i64, Vec<i64>, OffsetDateTime, Option<OffsetDateTime>);
|
||||
|
||||
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<T>(vec: &mut Vec<T>, 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<Option<Self>, 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::<<Call 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),
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
/// 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<Option<Self>, 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<Option<Self>, 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::<Vec<_>>();
|
||||
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<i64>)>()
|
||||
.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::<Vec<_>>();
|
||||
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(())
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user