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"); } }