From 2f99b32aea88981ffb0ff61f9ac286822d7cb262 Mon Sep 17 00:00:00 2001 From: Rafael Bradley Date: Mon, 18 Aug 2025 15:31:22 -0300 Subject: [PATCH] Final migration updates --- .dockerignore | 1 + .gitignore | 3 +- Cargo.lock | 23 ++ Cargo.toml | 3 + src/bin/migrate.rs | 432 ++++++++++++++++++++++++++ src/database/migrations/0001_init.cql | 3 +- src/database/models.rs | 159 ++++++++-- src/groups/calls/events.rs | 6 +- src/lib.rs | 9 + 9 files changed, 605 insertions(+), 34 deletions(-) create mode 100644 src/bin/migrate.rs create mode 100644 src/lib.rs diff --git a/.dockerignore b/.dockerignore index 073154f..01519b0 100644 --- a/.dockerignore +++ b/.dockerignore @@ -4,6 +4,7 @@ /.vscode /k8s /playbooks +/backup .gitignore .gitlab-ci.yml diff --git a/.gitignore b/.gitignore index 33fd1ec..8fc9422 100644 --- a/.gitignore +++ b/.gitignore @@ -1,9 +1,10 @@ /target /.venv /.ansible +/backup .env 00-vault-pwd.txt -reset-ssh.sh +reset-ssh.sh \ No newline at end of file diff --git a/Cargo.lock b/Cargo.lock index da2c637..1196bd6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -397,6 +397,27 @@ version = "0.8.21" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d0a5c400df2834b80a4c3327b3aad3a4c4cd4de0629063962b03235697506a28" +[[package]] +name = "csv" +version = "1.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "acdc4883a9c96732e4733212c01447ebd805833b7275a73ca3ee080fd77afdaf" +dependencies = [ + "csv-core", + "itoa", + "ryu", + "serde", +] + +[[package]] +name = "csv-core" +version = "0.1.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7d02f3b0da4c6504f86e9cd789d8dbafab48c2321be74e9987593de5a894d93d" +dependencies = [ + "memchr", +] + [[package]] name = "darling" version = "0.20.11" @@ -1223,6 +1244,7 @@ dependencies = [ "axum", "clap", "colored", + "csv", "dotenvy", "include_dir", "itertools", @@ -1236,6 +1258,7 @@ dependencies = [ "regex", "scc", "scylla", + "serde", "thiserror 2.0.12", "time", "tokio", diff --git a/Cargo.toml b/Cargo.toml index f714f57..0246b7f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -2,12 +2,14 @@ name = "payphone" version = "4.0.2" edition = "2024" +default-run = "payphone" [dependencies] async-trait = "0.1.88" axum = "0.8.4" clap = { version = "4.5.42", features = ["derive", "env", "string"] } colored = "3.0.0" +csv = "1.3.1" dotenvy = "0.15.7" include_dir = "0.7.4" itertools = "0.14.0" @@ -21,6 +23,7 @@ rand_chacha = { version = "0.9.0", features = ["os_rng"] } regex = "1.11.1" scc = "2.3.4" scylla = { version = "1.3.1", features = ["time-03"] } +serde = { version = "1.0.219", features = ["derive"] } thiserror = "2.0.12" time = { version = "0.3.41", features = ["formatting", "local-offset", "macros"] } tokio = { version = "1.47.1", features = ["macros", "rt-multi-thread"] } diff --git a/src/bin/migrate.rs b/src/bin/migrate.rs new file mode 100644 index 0000000..45b16d5 --- /dev/null +++ b/src/bin/migrate.rs @@ -0,0 +1,432 @@ +use std::{collections::HashMap, sync::Arc}; + +use clap::Parser; +use payphone::{ + database::{ + self, + models::{ + MessageData, Report as PayphoneReport, ReportResolution, Server as PayphoneServer, + UserIncident, UserIncidentType, + }, + }, + utils::{Bye, get_random_u64, u64_to_i64}, +}; +use serde::{Deserialize, Serialize}; +use time::OffsetDateTime; +use tokio::{sync::Semaphore, time::Instant}; + +#[derive(Parser)] +struct Args { + dir: String, + + #[arg(long, default_value_t = String::from("127.0.0.1"))] + database_addr: String, + + #[arg(long)] + database_username: Option, + + #[arg(long)] + database_password: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +struct Report { + id: u64, + message_id: u64, + message_content: String, + channel_id: u64, + user_id: u64, + created_at: String, + resolved: String, + resolved_at: Option, + resolved_by_id: Option, + user_avatar_url: String, + user_name: String, + reply_user_name: Option, + reply_message_content: Option, + reply_user_avatar_url: Option, + reply_user_id: Option, + reported_by_id: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +struct User { + id: u64, + is_masked: String, + is_moderator: String, + is_supporter_until: String, + warns_count: u64, + bans_count: u64, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +struct Ban { + id: u64, + user_id: u64, + created_at: String, + expires_at: Option, + created_by_id: u64, + expired_by_id: Option, + report_id: Option, + reason: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +struct Server { + id: u64, + prefix: String, + language: String, + name: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +struct Warn { + id: u64, + user_id: u64, + reason: String, + created_at: String, + created_by_id: u64, + report_id: u64, +} + +#[tokio::main] +async fn main() { + let args = Args::parse(); + + println!("Testing DB connection..."); + + let db = Arc::new( + database::init( + args.database_addr, + args.database_username, + args.database_password, + ) + .await + .expect("Failed to initialize database"), + ); + + println!("Reading migration file {}/reports.csv", args.dir); + + let mut rdr = csv::Reader::from_path(format!("{}/reports.csv", args.dir)) + .expect("Failed to open reports file"); + + let mut reports = Vec::new(); + + let format = time::format_description::parse( + "[year]-[month]-[day] [hour]:[minute]:[second].[subsecond][offset_hour]", + ) + .expect("Failed to parse format description"); + + for result in rdr.deserialize() { + let report: Report = result.expect("Failed to deserialize report"); + reports.push(report); + } + + println!("Finished reading reports, found {} reports", reports.len()); + println!("Reading migration file {}/users.csv", args.dir); + + let mut rdr = csv::Reader::from_path(format!("{}/users.csv", args.dir)) + .expect("Failed to open users file"); + + let mut users = HashMap::new(); + + for result in rdr.deserialize() { + let user: User = result.expect("Failed to deserialize user"); + users.insert(user.id, user).bye(); + } + + println!("Finished reading users, found {} users", users.len()); + println!("Reading migration file {}/bans.csv", args.dir); + + let mut rdr = + csv::Reader::from_path(format!("{}/bans.csv", args.dir)).expect("Failed to open bans file"); + + let mut bans = HashMap::new(); + + for result in rdr.deserialize() { + let ban: Ban = result.expect("Failed to deserialize ban"); + bans.insert(ban.id, ban).bye(); + } + + println!("Finished reading bans, found {} bans", bans.len()); + println!("Reading migration file {}/servers.csv", args.dir); + + let mut rdr = csv::Reader::from_path(format!("{}/servers.csv", args.dir)) + .expect("Failed to open servers file"); + + let mut servers = HashMap::new(); + + for result in rdr.deserialize() { + let server: Server = result.expect("Failed to deserialize server"); + servers.insert(server.id, server).bye(); + } + + println!("Finished reading servers, found {} servers", servers.len()); + println!("Reading migration file {}/warns.csv", args.dir); + + let mut rdr = csv::Reader::from_path(format!("{}/warns.csv", args.dir)) + .expect("Failed to open warns file"); + + let mut warns = HashMap::new(); + + for result in rdr.deserialize() { + let warn: Warn = result.expect("Failed to deserialize warn"); + warns.insert(warn.id, warn).bye(); + } + + println!("Finished reading warns, found {} warns", warns.len()); + println!("Starting migration..."); + println!("Inserting server configurations..."); + + let semaphore = Arc::new(Semaphore::new(50)); + let mut tasks = Vec::new(); + + for (i, server) in servers.values().enumerate() { + let payphone_server = PayphoneServer { + id: server.id, + prefix: server.prefix.clone(), + language: server.language.clone(), + name: server.name.clone(), + }; + + let task_db = db.clone(); + let task_semaphore = semaphore.clone(); + + tasks.push(tokio::spawn(async move { + let _permit = task_semaphore.acquire().await; + + let start = Instant::now(); + payphone_server + .insert(&task_db) + .await + .expect("Failed to insert server"); + + let duration = start.elapsed(); + println!( + "({i}) Inserted server {} in {:?}", + payphone_server.id, duration + ); + })); + } + + let bans = Arc::new(bans); + let warns = Arc::new(warns); + let reports = Arc::new(reports); + + for (i, report) in reports.iter().enumerate() { + let task_db = db.clone(); + let task_semaphore = semaphore.clone(); + let task_report = report.clone(); + let task_bans = bans.clone(); + let task_warns = warns.clone(); + let task_format = format.clone(); + + tasks.push(tokio::spawn(async move { + let _permit = task_semaphore.acquire().await; + + let start = Instant::now(); + + let resolution = if task_report.resolved == "t" { + if task_bans.iter().any(|(_id, ban)| { + ban.report_id + .map(|report_id| task_report.id == report_id) + .unwrap_or(false) + }) { + Some(ReportResolution::Ban) + } else if task_warns + .iter() + .any(|(_id, warn)| warn.report_id == task_report.id) + { + Some(ReportResolution::Warn) + } else { + Some(ReportResolution::Skip) + } + } else { + None + }; + + let payphone_report = PayphoneReport { + id: task_report.id, + user_id: task_report.user_id, + message: MessageData { + id: u64_to_i64(task_report.message_id), + call_id: 0, + author_id: u64_to_i64(task_report.user_id), + author_name: task_report.user_name, + author_avatar_url: task_report.user_avatar_url, + author_mask: None, + content: task_report.message_content, + attachments: vec![], + sticker: None, + propagation_message_id: None, + propagation_channel_id: None, + replying_to_message_id: None, + is_deleted: false, + is_edited: false, + is_censored: false, + }, + created_at: OffsetDateTime::parse(&task_report.created_at, &task_format) + .expect("Failed to parse created_at") + .to_utc(), + created_by_ids: task_report + .reported_by_id + .map(|id| vec![id]) + .unwrap_or_default(), + replying_to: None, + reviewed_at: task_report.resolved_at.as_ref().map(|s| { + OffsetDateTime::parse(s, &task_format) + .expect("Failed to parse resolved_at") + .to_utc() + }), + resolution, + reviewed_by_id: task_report.resolved_by_id, + }; + + payphone_report + .insert(&task_db) + .await + .expect("Failed to insert report"); + + let duration = start.elapsed(); + println!( + "({i}) Inserted report {} in {:?}", + payphone_report.id, duration + ); + })); + } + + for (i, ban) in bans.values().enumerate() { + let task_semaphore = semaphore.clone(); + let task_format = format.clone(); + let task_db = db.clone(); + let task_reports = reports.clone(); + let task_ban = Arc::new(ban.clone()); + + tasks.push(tokio::spawn(async move { + let _permit = task_semaphore.acquire().await; + + let start = Instant::now(); + + let message = task_reports + .iter() + .find(|report| { + task_ban + .report_id + .map(|id| id == report.id) + .unwrap_or(false) + }) + .map(|report| MessageData { + id: u64_to_i64(report.message_id), + call_id: 0, + author_id: u64_to_i64(report.user_id), + author_name: report.user_name.clone(), + author_avatar_url: report.user_avatar_url.clone(), + author_mask: None, + content: report.message_content.clone(), + attachments: vec![], + sticker: None, + propagation_message_id: None, + propagation_channel_id: None, + replying_to_message_id: None, + is_deleted: false, + is_edited: false, + is_censored: false, + }); + + let payphone_incident = UserIncident { + incident_id: get_random_u64(), + user_id: task_ban.user_id, + report_id: task_ban.report_id, + r#type: UserIncidentType::Ban, + message, + reason: task_ban.reason.clone(), + created_at: OffsetDateTime::parse(&task_ban.created_at, &task_format) + .expect("Failed to parse created_at") + .to_utc(), + created_by_id: Some(task_ban.created_by_id), + duration: None, + expires_at: task_ban.expires_at.as_ref().map(|expires_at| { + OffsetDateTime::parse(expires_at, &task_format) + .expect("Failed to parse expires_at") + .to_utc() + }), + }; + + payphone_incident + .insert(&task_db) + .await + .expect("Failed to insert ban"); + + let duration = start.elapsed(); + println!( + "({i}) Inserted incident (ban) {} in {:?}", + payphone_incident.incident_id, duration + ); + })); + } + + for (i, warn) in warns.values().enumerate() { + let task_semaphore = semaphore.clone(); + let task_format = format.clone(); + let task_db = db.clone(); + let task_reports = reports.clone(); + let task_warn = Arc::new(warn.clone()); + + tasks.push(tokio::spawn(async move { + let _permit = task_semaphore.acquire().await; + + let start = Instant::now(); + + let message = task_reports + .iter() + .find(|report| report.id == task_warn.report_id) + .map(|report| MessageData { + id: u64_to_i64(report.message_id), + call_id: 0, + author_id: u64_to_i64(report.user_id), + author_name: report.user_name.clone(), + author_avatar_url: report.user_avatar_url.clone(), + author_mask: None, + content: report.message_content.clone(), + attachments: vec![], + sticker: None, + propagation_message_id: None, + propagation_channel_id: None, + replying_to_message_id: None, + is_deleted: false, + is_edited: false, + is_censored: false, + }); + + let payphone_incident = UserIncident { + incident_id: get_random_u64(), + user_id: task_warn.user_id, + report_id: Some(task_warn.report_id), + r#type: UserIncidentType::Warning, + message, + reason: task_warn.reason.clone(), + created_at: OffsetDateTime::parse(&task_warn.created_at, &task_format) + .expect("Failed to parse created_at") + .to_utc(), + created_by_id: Some(task_warn.created_by_id), + duration: None, + expires_at: None, + }; + + payphone_incident + .insert(&task_db) + .await + .expect("Failed to insert warn"); + + let duration = start.elapsed(); + println!( + "({i}) Inserted incident (warn) {} in {:?}", + payphone_incident.incident_id, duration + ); + })); + } + + for task in tasks { + task.await.expect("Failed to join task"); + } +} diff --git a/src/database/migrations/0001_init.cql b/src/database/migrations/0001_init.cql index ee29466..e229477 100644 --- a/src/database/migrations/0001_init.cql +++ b/src/database/migrations/0001_init.cql @@ -76,7 +76,7 @@ CREATE TYPE IF NOT EXISTS payphone.message ( propagation_message_id BIGINT, propagation_channel_id BIGINT, - replying_to_id BIGINT, + replying_to_message_id BIGINT, is_deleted BOOLEAN, is_edited BOOLEAN, @@ -105,6 +105,7 @@ CREATE TABLE IF NOT EXISTS payphone.servers ( name TEXT, prefix TEXT, + language TEXT, ); CREATE TABLE IF NOT EXISTS payphone.reports ( diff --git a/src/database/models.rs b/src/database/models.rs index 52f9580..feaff9f 100644 --- a/src/database/models.rs +++ b/src/database/models.rs @@ -13,9 +13,7 @@ use thiserror::Error; use time::{Duration, OffsetDateTime, UtcDateTime, UtcOffset}; use twilight_http::Client; use twilight_model::{ - channel::Message as TwilightMessage, - guild::{Guild as TwilightGuild, GuildInfo as TwilightGuildInfo}, - id::Id, + channel::Message as TwilightMessage, guild::Guild as TwilightGuild, id::Id, user::User as TwilightUser, }; @@ -205,7 +203,7 @@ impl User { pub async fn incident( &self, reason: String, - message: MessageData, + message: Option, created_by_id: Option, session: &Session, ) -> Result { @@ -270,21 +268,28 @@ pub struct Server { pub id: u64, pub name: String, pub prefix: String, + pub language: String, } impl<'a> Object<'a> for Server { - type Row = (i64, String, String); + type Row = (i64, String, String, String); fn from_row(row: Self::Row) -> Self { Self { id: i64_to_u64(row.0), name: row.1, prefix: row.2, + language: row.3, } } fn to_row(value: Self) -> Self::Row { - (u64_to_i64(value.id), value.name, value.prefix) + ( + u64_to_i64(value.id), + value.name, + value.prefix, + value.language, + ) } } @@ -298,7 +303,8 @@ impl Server { SELECT server_id, name, - prefix + prefix, + language FROM payphone.servers WHERE @@ -360,8 +366,9 @@ impl Server { INSERT INTO payphone.servers ( server_id, name, - prefix - ) VALUES (?, ?, ?) IF NOT EXISTS; + prefix, + language + ) VALUES (?, ?, ?, ?) IF NOT EXISTS; "#, Self::to_row(self.clone()), ) @@ -379,11 +386,13 @@ impl Server { INSERT INTO payphone.servers ( server_id, name, - prefix + prefix, + language ) VALUES ( ?, ?, - ? + ?, + ? ); "#, Self::to_row(self.clone()), @@ -413,16 +422,7 @@ impl From for Server { 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(), + language: value.preferred_locale, } } } @@ -933,7 +933,7 @@ pub struct MessageData { pub propagation_message_id: Option, pub propagation_channel_id: Option, - pub replying_to_id: Option, + pub replying_to_message_id: Option, pub is_deleted: bool, pub is_edited: bool, @@ -952,7 +952,7 @@ pub struct MessageBuilder { is_censored: bool, propagation_message_id: Option, propagation_channel_id: Option, - replying_to_id: Option, + replying_to_message_id: Option, } impl MessageBuilder { @@ -964,7 +964,7 @@ impl MessageBuilder { is_censored: false, propagation_message_id: None, propagation_channel_id: None, - replying_to_id: None, + replying_to_message_id: None, } } @@ -1001,8 +1001,8 @@ impl MessageBuilder { // Argument is made optional here because it happens to be more convenient for where this is // used. - pub fn replying_to_id(mut self, replying_to_id: Option) -> Self { - self.replying_to_id = replying_to_id; + pub fn replying_to_message_id(mut self, replying_to_id: Option) -> Self { + self.replying_to_message_id = replying_to_id; self } @@ -1026,7 +1026,7 @@ impl MessageBuilder { .map(|s| u64_to_i64(s.id.get())), propagation_message_id: self.propagation_message_id.map(u64_to_i64), propagation_channel_id: self.propagation_channel_id.map(u64_to_i64), - replying_to_id: self.replying_to_id.map(u64_to_i64), + replying_to_message_id: self.replying_to_message_id.map(u64_to_i64), // You can't create a deleted message, this is always going to be false when // initializing @@ -1225,6 +1225,107 @@ impl MessageByPropagationId { } } +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum ReportResolution { + Ban, + Warn, + Skip, +} + +#[derive(Debug, Clone)] +pub struct Report { + pub id: u64, + pub user_id: u64, + + pub message: MessageData, + pub replying_to: Option, + + pub created_at: UtcDateTime, + pub reviewed_at: Option, + + pub created_by_ids: Vec, + + pub reviewed_by_id: Option, + pub resolution: Option, +} + +impl<'a> Object<'a> for Report { + type Row = ( + i64, + i64, + MessageData, + Option, + OffsetDateTime, + Option, + Vec, + Option, + Option, + ); + + fn from_row(row: Self::Row) -> Self { + Self { + id: i64_to_u64(row.0), + user_id: i64_to_u64(row.1), + message: row.2, + replying_to: row.3, + created_at: row.4.to_utc(), + reviewed_at: row.5.map(|i| i.to_utc()), + created_by_ids: row.6.into_iter().map(i64_to_u64).collect(), + reviewed_by_id: row.7.map(i64_to_u64), + resolution: row.8.map(|r| match r.as_str() { + "ban" => ReportResolution::Ban, + "warn" => ReportResolution::Warn, + "skip" => ReportResolution::Skip, + _ => unreachable!(), + }), + } + } + + fn to_row(value: Self) -> Self::Row { + ( + u64_to_i64(value.id), + u64_to_i64(value.user_id), + value.message, + value.replying_to, + value.created_at.to_offset(UtcOffset::UTC), + value.reviewed_at.map(|i| i.to_offset(UtcOffset::UTC)), + value.created_by_ids.into_iter().map(u64_to_i64).collect(), + value.reviewed_by_id.map(u64_to_i64), + value.resolution.map(|r| match r { + ReportResolution::Ban => "ban".into(), + ReportResolution::Warn => "warn".into(), + ReportResolution::Skip => "skip".into(), + }), + ) + } +} + +impl Report { + pub async fn insert(&self, session: &Session) -> Result<(), DatabaseError> { + session + .query_unpaged( + r#" + INSERT INTO payphone.reports ( + report_id, + user_id, + message, + replying_to, + created_at, + reviewed_at, + created_by_ids, + reviewed_by_id, + resolution + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?); + "#, + Self::to_row(self.clone()), + ) + .await + .map_err(DatabaseError::QueryError)?; + + Ok(()) + } +} + #[derive(Debug, Clone, PartialEq, Eq)] pub enum UserIncidentType { Warning, @@ -1239,7 +1340,7 @@ pub struct UserIncident { pub r#type: UserIncidentType, - pub message: MessageData, + pub message: Option, pub reason: String, @@ -1256,7 +1357,7 @@ impl<'a> Object<'a> for UserIncident { i64, Option, String, - MessageData, + Option, String, Option, OffsetDateTime, diff --git a/src/groups/calls/events.rs b/src/groups/calls/events.rs index 4d4d941..db5c973 100644 --- a/src/groups/calls/events.rs +++ b/src/groups/calls/events.rs @@ -45,7 +45,7 @@ async fn penalize_user( let new_incident = user .incident( reason.clone(), - message.data.clone(), + Some(message.data.clone()), Some(user.id), &rt.state.db, ) @@ -219,7 +219,7 @@ impl OnMessageHandler for CallMessageHandler { .message(event.0.clone()) .profile(profile.clone()) .is_censored(!rules.is_empty()) - .replying_to_id(replying_to_id) + .replying_to_message_id(replying_to_id) .build() .expect( "The message builder was missing either the call ID, the message, or the profile", @@ -266,7 +266,7 @@ impl OnMessageHandler for CallMessageHandler { .message(event.0.clone()) .profile(profile.clone()) .is_censored(!rules.is_empty()) - .replying_to_id(replying_to_id) + .replying_to_message_id(replying_to_id) .propagation_message_id(message_sent.id.get()) .propagation_channel_id(opposite_channel.id) .build() diff --git a/src/lib.rs b/src/lib.rs new file mode 100644 index 0000000..be460e2 --- /dev/null +++ b/src/lib.rs @@ -0,0 +1,9 @@ +pub mod cli; +pub mod constants; +pub mod database; +pub mod discord; +pub mod framework; +pub mod groups; +pub mod logging; +pub mod state; +pub mod utils;