init

src/services/emoji.rs

@@ -1,10 +1,11 @@
 use crate::http::error::HTTPError;
 use crate::models::emoji;
 use crate::services::ServicesContext;
+use crate::services::media::PendingMediaFile;
 use sea_orm::{ActiveModelTrait, Set, TransactionTrait};
 use sha2::{Digest, Sha256};
 use std::{
-    path::{Path, PathBuf},
+    path::Path,
     sync::Arc,
 };
 use tokio::fs;
@@ -85,17 +86,18 @@ impl EmojiService {
         root: &Path,
         id: Uuid,
         data: &[u8],
+        extension: Option<&str>,
     ) -> Result<(String, String), HTTPError> {
-        let dir = root.join("emoji");
-        fs::create_dir_all(&dir)
+        let mut pending = PendingMediaFile::begin(root, "emoji", id, extension)
             .await
             .map_err(|e| HTTPError::InternalServerError(e.to_string()))?;
-        let relative = PathBuf::from("emoji").join(id.to_string());
-        let path = root.join(&relative);
-        fs::write(&path, data)
+        pending.write(data)
             .await
             .map_err(|e| HTTPError::InternalServerError(e.to_string()))?;
-        Ok((relative.to_string_lossy().into_owned(), Self::hash(data)))
+        let relative = pending.finish()
+            .await
+            .map_err(|e| HTTPError::InternalServerError(e.to_string()))?;
+        Ok((relative, Self::hash(data)))
     }
     pub async fn remove_asset(root: &Path, path: Option<&str>) {
         if let Some(path) = path {

src/services/media.rs

@@ -0,0 +1,132 @@
+use std::path::{Path, PathBuf};
+use std::time::{Duration, SystemTime};
+use tokio::fs::{self, File};
+use tokio::io::AsyncWriteExt;
+use uuid::Uuid;
+
+pub struct PendingMediaFile {
+    file: File,
+    temporary_path: PathBuf,
+    final_path: PathBuf,
+    relative_final_path: String,
+}
+
+impl PendingMediaFile {
+    pub async fn begin(root: &Path, directory: &str, id: Uuid, extension: Option<&str>) -> std::io::Result<Self> {
+        let relative_directory = PathBuf::from(directory);
+        let directory_path = root.join(&relative_directory);
+        fs::create_dir_all(&directory_path).await?;
+
+        let uuid_name = id.to_string();
+        let temporary_name = format!("~{uuid_name}.part");
+        let final_name = match extension.filter(|value| !value.is_empty()) {
+            Some(extension) => format!("{uuid_name}.{extension}"),
+            None => uuid_name,
+        };
+        let temporary_path = directory_path.join(temporary_name);
+        let final_path = directory_path.join(&final_name);
+        let file = File::create(&temporary_path).await?;
+
+        Ok(Self {
+            file,
+            temporary_path,
+            final_path,
+            relative_final_path: relative_directory.join(final_name).to_string_lossy().into_owned(),
+        })
+    }
+
+    pub async fn write(&mut self, chunk: &[u8]) -> std::io::Result<()> {
+        self.file.write_all(chunk).await
+    }
+
+    pub async fn finish(mut self) -> std::io::Result<String> {
+        self.file.flush().await?;
+        self.file.sync_all().await?;
+        drop(self.file);
+        fs::rename(&self.temporary_path, &self.final_path).await?;
+        Ok(self.relative_final_path)
+    }
+
+    pub async fn remove_final(path: &Path) {
+        let _ = fs::remove_file(path).await;
+    }
+}
+
+pub fn extension_from_filename(filename: &str) -> Option<String> {
+    let name = Path::new(filename).file_name()?.to_str()?;
+    let lower = name.to_ascii_lowercase();
+    let extension = ["tar.gz", "tar.bz2", "tar.xz", "tar.zst"]
+        .iter()
+        .find(|candidate| lower.ends_with(&format!(".{candidate}")))
+        .map(|candidate| (*candidate).to_string())
+        .or_else(|| Path::new(name).extension().and_then(|value| value.to_str()).map(str::to_ascii_lowercase))?;
+    if extension.chars().all(|character| character.is_ascii_alphanumeric() || character == '.') {
+        Some(extension)
+    } else {
+        None
+    }
+}
+
+pub fn extension_from_mime(mime_type: &str) -> Option<&'static str> {
+    match mime_type {
+        "image/png" => Some("png"),
+        "image/gif" => Some("gif"),
+        "image/webp" => Some("webp"),
+        "image/jpeg" => Some("jpg"),
+        _ => None,
+    }
+}
+
+pub async fn cleanup_temporary_files(root: &Path, max_age: Duration) -> std::io::Result<()> {
+    let mut directories = vec![root.to_path_buf()];
+    let mut root_entries = match fs::read_dir(root).await {
+        Ok(entries) => entries,
+        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(()),
+        Err(error) => return Err(error),
+    };
+    while let Some(entry) = root_entries.next_entry().await? {
+        if entry.file_type().await?.is_dir() {
+            directories.push(entry.path());
+        }
+    }
+
+    let now = SystemTime::now();
+    for directory in directories {
+        let mut entries = match fs::read_dir(&directory).await {
+            Ok(entries) => entries,
+            Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue,
+            Err(error) => return Err(error),
+        };
+        while let Some(entry) = entries.next_entry().await? {
+            let name = entry.file_name().to_string_lossy().into_owned();
+            if !name.starts_with('~') || !name.ends_with(".part") {
+                continue;
+            }
+            let modified = entry.metadata().await?.modified().unwrap_or(now);
+            if now.duration_since(modified).unwrap_or_default() > max_age {
+                let _ = fs::remove_file(entry.path()).await;
+            }
+        }
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::extension_from_filename;
+
+    #[test]
+    fn preserves_compound_extensions() {
+        assert_eq!(extension_from_filename("archive.tar.gz").as_deref(), Some("tar.gz"));
+    }
+
+    #[test]
+    fn extracts_simple_extensions() {
+        assert_eq!(extension_from_filename("photo.PNG").as_deref(), Some("png"));
+    }
+
+    #[test]
+    fn returns_none_without_an_extension() {
+        assert_eq!(extension_from_filename("README").as_deref(), None);
+    }
+}

src/services/message.rs

@@ -1,10 +1,10 @@
 use crate::domain::events::message::{
     MessageCreatedEvent, MessageDeletedEvent, MessageUpdatedEvent,
 };
-use crate::models::{channel, message};
+use crate::models::{attachment, channel, message};
 use crate::services::ServicesContext;
 use event_bus::Scope;
-use sea_orm::{ActiveModelTrait, EntityTrait, QuerySelect, Set, TransactionTrait};
+use sea_orm::{ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, QuerySelect, Set, TransactionTrait};
 use std::sync::Arc;
 use uuid::Uuid;
 
@@ -67,6 +67,51 @@ impl MessageService {
         Ok(msg)
     }
 
+    pub async fn create_message_with_attachments(
+        &self,
+        channel_id: Uuid,
+        author_id: Uuid,
+        content: String,
+        file_ids: Vec<Uuid>,
+        reply_to_id: Option<Uuid>,
+    ) -> Result<message::Model, anyhow::Error> {
+        let db = &self.service_context.repositories.server.context.db;
+        let event_bus = &self.service_context.event_bus;
+        let txn = db.begin().await?;
+        let files = if file_ids.is_empty() {
+            Vec::new()
+        } else {
+            attachment::Entity::find()
+                .filter(attachment::Column::Id.is_in(file_ids.clone()))
+                .all(&txn)
+                .await?
+        };
+        if files.len() != file_ids.len()
+            || files.iter().any(|file| file.channel_id != channel_id || file.user_id != author_id || file.message_id.is_some())
+        {
+            return Err(anyhow::anyhow!("Invalid or already used attachment"));
+        }
+        let msg = message::ActiveModel {
+            channel_id: Set(channel_id), user_id: Set(author_id), content: Set(content),
+            reply_to_id: Set(reply_to_id), ..Default::default()
+        }.insert(&txn).await?;
+        for file in files {
+            let mut active: attachment::ActiveModel = file.into();
+            active.message_id = Set(Some(msg.id));
+            active.update(&txn).await?;
+        }
+        txn.commit().await?;
+        let server_id = channel::Entity::find_by_id(msg.channel_id)
+            .select_only().column(channel::Column::ServerId)
+            .into_tuple::<Option<Uuid>>().one(db).await?.flatten();
+        let mut scopes = vec![Scope::uuid("channel", msg.channel_id)];
+        if let Some(server_id) = server_id { scopes.push(Scope::uuid("server", server_id)); }
+        event_bus.emit_scoped("message_created", scopes, MessageCreatedEvent {
+            server_id, channel_id: msg.channel_id, message: msg.clone(),
+        });
+        Ok(msg)
+    }
+
     pub async fn update_message(
         &self,
         id: Uuid,

src/services/mod.rs

@@ -18,6 +18,7 @@ pub mod channel;
 pub mod emoji;
 pub mod message;
 pub mod message_reaction;
+pub mod media;
 mod permission;
 pub mod permission_sync;
 pub mod realtime_registry;