use std::time::{SystemTime, UNIX_EPOCH}; use tokio::fs; use tokio::io::AsyncWriteExt; use mysql_async::params; use mysql_async::Pool; use mysql_async::prelude::Queryable; use mysql_async::Error as SqlError; use crate::Response; use crate::{UBigInt, BigInt}; use crate::{Channel, Message, UserMessage}; use crate::common; use rand::RngCore; const MAX_MESSAGES: u64 = 1000; impl Message { fn new(id: u64, time: i64, content: &str, content_type: &str, uid: u64, cid: u64) -> Self { // Here for convenience pretty straight forward, copying at the last second Message { id, time, content: content.into(), content_type: content_type.into(), author_id: uid, channel_id: cid } } async fn insert_text(p: &Pool, content: &str, cid: UBigInt, uid: UBigInt) -> Result, SqlError> { // TODO: make this check not compare against something so hardcoded and if content.len() > 4_000 { return Ok(Response::RestrictedInput("Text larger than 4000 bytes".into())) } else { let mut conn = p.get_conn().await?; let q = "INSERT INTO messages (id, time, content, content_type, author_id, channel_id) VALUES(:id, :time, :content, 'text/plain', :uid, :chan_id)"; let id: u64 = rand::rngs::OsRng.next_u64(); let now: i64 = SystemTime::now() .duration_since(UNIX_EPOCH) .expect("System time `NOW` failed") .as_millis() as i64; let p = params!{ "id" => id, "time" => now, "content" => content, "uid" => uid, "chan_id" => cid }; conn.exec_drop(q, p).await?; let msg = Message::new(id, now, content, "text/plain", uid, cid); return Ok(Response::Row(msg)) } } async fn save_file(filename: String, data: &[u8], msg: Message) -> Response { match fs::File::create(filename).await { Ok(mut file) => { if let Ok(_) = file.write_all(data).await { Response::Row(msg) } else { Response::Other(format!("Couldn't write data to disk")) } } Err(e) => { let msg = format!("db::Message::save_file {}", e); Response::Other(msg) } } } pub async fn send(p: &Pool, content: &str, content_type: &str, cid: UBigInt, uid: UBigInt) -> Result, SqlError> { //! @returns on_sucess -> Ok(Response::Row) //! @returns on_user_fail -> Ok(Response::RestrictedInput) //! @returns on_caught_server_fail -> Ok(Response::Other) //! @returns on_failure Err(SqlErr) if content_type == "text/plain" { match Self::insert_text(p, content, cid, uid).await { Ok(pass) => Ok(pass), Err(e) => common::try_error_passthrough(e) } } else { if content.len() > 10_000_000 { return Ok(Response::RestrictedInput("Large data not allowed".into())) } let extension = match content_type { "image/png" => "png", "image/jpeg" | "image/jpg" => "jpg", "application/webm" => "webm", "application/mp4" => "mp4", "application/mp3" => "mp3", _ => panic!("Bad file type sent to db layer {}", content_type) }; let q = "INSERT INTO messages (id, time, content, content_type, author_id, channel_id) VALUES(:id, :time, :fname, :ctype, :author, :channel)"; let id: u64 = rand::rngs::OsRng.next_u64(); let filename = format!("{}.{}", id, extension); let now: i64 = SystemTime::now() .duration_since(UNIX_EPOCH) .expect("System time `NOW` failed") .as_millis() as i64; let mut conn = p.get_conn().await?; let insert_res = conn.exec_drop(q, params!{ "id" => id, "time" => now, "ctype" => content_type, "fname" => &filename, "author" => uid, "channel" => cid }).await; match insert_res { Ok(_) => { let msg = Message::new(id, now, &filename, content_type, uid, cid); Ok(Message::save_file(filename, content.as_bytes(), msg).await) }, Err(e) => common::try_error_passthrough(e) } } } pub async fn get_time_range(p: &Pool, channel_id: UBigInt, start: BigInt, end: BigInt, limit: Option) -> Result, SqlError> { //! @returns on success : Set(Vec) //! @returns on userfail: RestrictedInput(message) //! @returns on error : Err(SqlError) if start >= end { Ok(Response::RestrictedInput("Invalid start/end parameters".into())) } else { let mut conn = p.get_conn().await?; let limit = if let Some(limit) = limit { match limit { 1 ..= MAX_MESSAGES => limit, _ => MAX_MESSAGES } } else { MAX_MESSAGES }; let q = " SELECT mem.name, msg.id, msg.time, msg.content, msg.content_type, msg.author_id FROM messages as msg JOIN members as mem ON mem.id = msg.author_id WHERE channel_id = :channel AND time >= :start AND time < :end LIMIT :limit"; let params = params!{ "start" => start, "end" => end, "channel" => channel_id, "limit" => limit }; let messages = conn.exec_map(q, params, |(name, id, time, content, content_type, author_id)| { UserMessage { id, time, content, content_type, author_id, name, channel_id } }).await?; Ok(Response::Set(messages)) } } pub async fn get_from_id(p: &Pool, channel_id: UBigInt, start: UBigInt, limit: Option) -> Result, SqlError> { //! @returns on success : Set(Vec) //! @returns on user failure : RestrictedInput(String) //! @returns on failure : Err(SqlError) let mut conn = p.get_conn().await?; let limit = if let Some(limit) = limit{ match limit { 1 ..= MAX_MESSAGES => limit, _ => MAX_MESSAGES } } else { MAX_MESSAGES // messages at a time }; let q = "SELECT id, time, content, content_type, author_id FROM messages WHERE channel_id = :channel AND id >= :start LIMIT :limit"; let params = params!{ "channel" => channel_id, "start" => start, "limit" => limit }; let messages = conn.exec_map(q, params, |(id, time, content, content_type, author_id)| { Self { id, time, content, content_type, author_id, channel_id } }).await?; Ok(Response::Set(messages)) } /// On success Response::Set /// On user failure Response::RestrictedInput /// On error Err pub async fn last_n(p: &Pool, limit: i32, channel_id: u64) -> Result, SqlError> { if limit <= 0 || limit > 100 { return Ok(Response::RestrictedInput("Invalid \"limit\" value".into())) } // Explicitly return empty if the channel requested does not exist if let None = Channel::get(p, channel_id).await? { return Ok(Response::Empty) } // Reminder: if content_type is not text/plain then content will be empty let q = " SELECT mem.name, msg.id, msg.time, msg.content, msg.content_type, msg.author_id FROM messages as msg JOIN members as mem ON mem.id = msg.author_id WHERE channel_id = :channel ORDER BY id DESC LIMIT :limit"; let params = params!{"limit" => limit, "channel" => channel_id}; let mut conn = p.get_conn().await?; let messages = conn.exec_map(q, params, |(name, id, time, content, content_type, author_id)| { UserMessage { name, id, time, content, content_type, author_id, channel_id } }).await?; return Ok(Response::Set(messages)) } }