headless_lms_server/controllers/helpers/

file_uploading.rs

//! Helper functions related to uploading to file storage.

pub use crate::domain::authorization::AuthorizationToken;
use crate::prelude::*;
use actix_http::header::HeaderMap;
use actix_multipart as mp;
use actix_multipart::Field;
use actix_web::http::header;
use futures::{StreamExt, TryStreamExt};
use headless_lms_utils::file_store::{FileStore, GenericPayload};
use headless_lms_utils::{
    file_store::file_utils::get_extension_from_filename, strings::generate_random_string,
};
use models::organizations::DatabaseOrganization;
use rand::distr::Alphanumeric;
use rand::distr::SampleString;
use std::{collections::HashSet, path::Path};
use std::{
    path::PathBuf,
    sync::{
        Arc, Mutex,
        atomic::{AtomicU64, Ordering},
    },
};
use utoipa::ToSchema;

const EXERCISE_UPLOAD_MAX_FILES: usize = 10;
const EXERCISE_UPLOAD_MAX_FILE_BYTES: u64 = 100 * 1024 * 1024;
const EXERCISE_UPLOAD_MAX_BATCH_BYTES: u64 = 100 * 1024 * 1024;

/// What an upload route returns for one stored file: the `file_uploads` row id an answer names it
/// by, and the URL it can be fetched from.
#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct ExerciseServiceUploadResultEntry {
    pub id: Uuid,
    pub url: String,
}

/// One stored upload, with the file details the iframe's wire result omits and the client API's
/// response carries.
pub struct ExerciseServiceUpload {
    pub entry: ExerciseServiceUploadResultEntry,
    pub name: String,
    pub mime: String,
    pub size_bytes: i64,
}

/** Tracks uploaded object paths for cleanup when the batch fails. */
pub struct ExerciseServiceUploadCleanup {
    pub path: String,
}

/// Deletes the objects an upload has already stored, unless the upload got far enough to
/// [`disarm`](Self::disarm) it.
///
/// Objects reach the store before their rows are committed, so an upload that fails part way has to
/// delete them itself; the reaper only finds objects through the rows that were never written. On
/// the handler's own error path call [`clean_up`](Self::clean_up) and await it. The `Drop` path is
/// the backstop for the case that has no error path at all: the client aborting the multipart body,
/// which drops the handler future at an await point.
pub struct UploadCleanup {
    pub uploaded_paths: Vec<ExerciseServiceUploadCleanup>,
    file_store: web::Data<dyn FileStore>,
    armed: bool,
}

impl UploadCleanup {
    pub fn new(file_store: web::Data<dyn FileStore>) -> Self {
        Self {
            uploaded_paths: Vec::new(),
            file_store,
            armed: true,
        }
    }

    /// Deletes what has been stored so far and disarms, so the `Drop` backstop cannot delete the
    /// same objects again.
    pub async fn clean_up(&mut self) {
        self.armed = false;
        for uploaded in std::mem::take(&mut self.uploaded_paths) {
            if let Err(delete_error) = self.file_store.delete(Path::new(&uploaded.path)).await {
                error!(
                    "Failed to delete file '{}' during cleanup: {delete_error}",
                    uploaded.path
                );
            }
        }
    }

    /// Call only once the uploads are recorded, with no await in between: an await would give the
    /// runtime a chance to drop the handler and delete objects that already have rows.
    pub fn disarm(&mut self) {
        self.armed = false;
    }
}

impl Drop for UploadCleanup {
    fn drop(&mut self) {
        if !self.armed || self.uploaded_paths.is_empty() {
            return;
        }
        let uploaded_paths = std::mem::take(&mut self.uploaded_paths);
        let file_store = self.file_store.clone();
        // `drop` cannot await, so the deletes run detached and outlive this request.
        actix_web::rt::spawn(async move {
            for uploaded in uploaded_paths {
                if let Err(delete_error) = file_store.delete(Path::new(&uploaded.path)).await {
                    error!(
                        "Failed to delete file '{}' during cleanup: {delete_error}",
                        uploaded.path
                    );
                }
            }
        });
    }
}

struct ExerciseServiceUploadMetadata {
    file_upload_id: Uuid,
    path: String,
    filename: String,
    mime_type: String,
    size_bytes: i64,
    url: String,
}

/// Who an answer upload is bound to, and where its objects are stored.
pub struct AnswerUploadDestination {
    /// The course or exam the exercise belongs to, which the object path is filed under.
    pub owner: CourseOrExamId,
    pub exercise_id: Uuid,
    pub user_id: Uuid,
}

/// Where an upload's objects are stored, and what the upload route hands back for them.
pub enum UploadPathScheme<'a> {
    /// `<exercise service slug>/<random>` and a permanent url naming that path, for files that
    /// belong to course content rather than to one student: a teacher's spec files, and the
    /// playground's throwaway uploads. The slug carries no authorization meaning.
    ExerciseService { exercise_service_slug: &'a str },
    /// `answers/v1/<course|exam>/<owner>/exercise/<exercise>/user/<user>/<file upload>`, so a
    /// retention or takedown rule can find every object of one course, exercise or student without
    /// reading the database, and every object leads back to its row.
    ///
    /// The path names the student, so the url handed back carries a claim instead of the path.
    Answer(&'a AnswerUploadDestination),
}

impl UploadPathScheme<'_> {
    /// Where one file of this upload is stored. Applies to new uploads only: old rows keep the
    /// layout they were written under, since nothing derives a path from a row.
    fn path(&self, file_upload_id: Uuid) -> String {
        match self {
            Self::ExerciseService {
                exercise_service_slug,
            } => format!("{exercise_service_slug}/{}", generate_random_string(32)),
            Self::Answer(AnswerUploadDestination {
                owner,
                exercise_id,
                user_id,
            }) => {
                // No fallback arm: a third kind of owner has to fail to compile here rather
                // than land under a path no cleanup rule knows about.
                let (owner_kind, owner_id) = match owner {
                    CourseOrExamId::Course(course_id) => ("course", course_id),
                    CourseOrExamId::Exam(exam_id) => ("exam", exam_id),
                };
                format!(
                    "answers/v1/{owner_kind}/{owner_id}/exercise/{exercise_id}/user/{user_id}/{file_upload_id}"
                )
            }
        }
    }

    /// The url the upload route hands back for one stored file.
    fn download_url(
        &self,
        file_upload_id: Uuid,
        path: &str,
        file_store: &dyn FileStore,
        app_conf: &ApplicationConfiguration,
    ) -> Result<String, ControllerError> {
        match self {
            Self::ExerciseService { .. } => {
                Ok(format!("{}/api/v0/files/{path}", app_conf.base_url))
            }
            Self::Answer(_) => Ok(file_store.get_claimed_download_url(file_upload_id, app_conf)?),
        }
    }
}

/// Processes a spec-file upload from an exercise service or the playground.
/// This function assumes that any permission checks have already been made.
///
/// `exercise_service_slug` namespaces the stored objects and is recorded with the upload; it
/// carries no authorization meaning. The playground passes its reserved `playground` slug, which
/// names no exercise service.
pub async fn process_exercise_service_upload(
    conn: &mut PgConnection,
    exercise_service_slug: &str,
    payload: Multipart,
    file_store: &dyn FileStore,
    uploaded_paths: &mut Vec<ExerciseServiceUploadCleanup>,
    uploader: Option<Uuid>,
    app_conf: &ApplicationConfiguration,
) -> Result<Vec<ExerciseServiceUpload>, ControllerError> {
    let streamed = stream_exercise_service_upload(
        UploadPathScheme::ExerciseService {
            exercise_service_slug,
        },
        payload,
        file_store,
        uploaded_paths,
        app_conf,
    )
    .await?;
    let mut tx = conn.begin().await?;
    let uploads = record_exercise_service_upload(&mut tx, streamed, uploader).await?;
    let file_upload_ids: Vec<Uuid> = uploads.iter().map(|upload| upload.entry.id).collect();
    // Recorded in the same transaction as the file rows: an upload the reaper cannot see is an
    // upload nothing will ever reclaim, since a spec's references are invisible to the host.
    models::exercise_spec_uploads::insert_many(
        &mut tx,
        exercise_service_slug,
        uploader,
        &file_upload_ids,
    )
    .await?;
    tx.commit().await?;
    Ok(uploads)
}

/// The parts of an upload that have reached the object store but have no database row yet.
pub struct StreamedExerciseServiceUpload {
    parts: Vec<ExerciseServiceUploadMetadata>,
}

/// Streams every multipart part to the object store, issuing no statements at all.
///
/// Split from the row inserts on purpose: the limits here are byte-based, not time-based, so a
/// handler that opened a transaction first would hold a connection `idle in transaction` for as
/// long as the client cares to trickle 100 MiB, and enough concurrent slow uploads would exhaust
/// the pool and hold back the vacuum xmin horizon.
pub async fn stream_exercise_service_upload(
    path_scheme: UploadPathScheme<'_>,
    mut payload: Multipart,
    file_store: &dyn FileStore,
    uploaded_paths: &mut Vec<ExerciseServiceUploadCleanup>,
    app_conf: &ApplicationConfiguration,
) -> Result<StreamedExerciseServiceUpload, ControllerError> {
    let mut parts = Vec::new();
    let mut ids = HashSet::new();
    let batch_bytes = Arc::new(AtomicU64::new(0));
    while let Some(item) = payload.next().await {
        let field = item.map_err(|err| {
            controller_err!(
                BadRequest,
                format!("Failed to read multipart field: {}", err),
                anyhow::anyhow!("Multipart error: {}", err)
            )
        })?;
        validate_exercise_upload_file_count(parts.len())?;
        let field_name = {
            let name_ref = field.name().ok_or_else(|| {
                controller_err!(
                    BadRequest,
                    "Tried to upload a multipart field without a field name or field ID"
                        .to_string()
                )
            })?;
            name_ref.to_string()
        };

        validate_exercise_upload_id(&field_name, &mut ids)?;
        let filename = validate_exercise_upload_filename(
            field
                .content_disposition()
                .and_then(|disposition| disposition.get_filename()),
        )?
        .to_string();

        // Minted here rather than by the insert: the object is streamed to the store before its
        // row exists, and a path may name the row it belongs to.
        let file_upload_id = Uuid::new_v4();
        let path = path_scheme.path(file_upload_id);
        uploaded_paths.push(ExerciseServiceUploadCleanup { path: path.clone() });
        let mime_type = field
            .content_type()
            .map(ToString::to_string)
            .unwrap_or_default();
        let (stream, stream_error, uploaded_bytes) =
            limited_exercise_upload_stream(field, batch_bytes.clone());
        let upload_result = file_store
            .upload_stream(Path::new(&path), stream, &mime_type)
            .await;
        if let Some(error) = stream_error
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .take()
        {
            return Err(error);
        }
        upload_result?;
        let url = path_scheme.download_url(file_upload_id, &path, file_store, app_conf)?;
        parts.push(ExerciseServiceUploadMetadata {
            file_upload_id,
            path,
            filename,
            mime_type,
            size_bytes: uploaded_bytes.load(Ordering::SeqCst) as i64,
            url,
        });
    }
    validate_exercise_upload_not_empty(parts.len())?;
    Ok(StreamedExerciseServiceUpload { parts })
}

/// Records the `file_uploads` rows for an already-streamed upload.
///
/// Takes the caller's transaction so that a caller which has further rows to write — the answer
/// upload routes bind each upload to an exercise and user — lands them together with these.
pub async fn record_exercise_service_upload(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    streamed: StreamedExerciseServiceUpload,
    uploader: Option<Uuid>,
) -> Result<Vec<ExerciseServiceUpload>, ControllerError> {
    let StreamedExerciseServiceUpload { parts } = streamed;
    let mut uploads = Vec::with_capacity(parts.len());
    for part in parts {
        models::file_uploads::insert_with_id(
            tx,
            part.file_upload_id,
            &part.filename,
            &part.path,
            &part.mime_type,
            uploader,
            Some(part.size_bytes),
        )
        .await?;
        uploads.push(ExerciseServiceUpload {
            entry: ExerciseServiceUploadResultEntry {
                id: part.file_upload_id,
                url: part.url,
            },
            name: part.filename,
            mime: part.mime_type,
            size_bytes: part.size_bytes,
        });
    }
    Ok(uploads)
}

/// Returns a side channel for typed upload-limit errors because the stream yields `anyhow` errors,
/// and the running byte count of this part. Both are only meaningful once `upload_stream` has
/// returned: the count is still climbing while the stream is being consumed.
fn limited_exercise_upload_stream(
    field: mp::Field,
    batch_bytes: Arc<AtomicU64>,
) -> (
    GenericPayload,
    Arc<Mutex<Option<ControllerError>>>,
    Arc<AtomicU64>,
) {
    let per_file_bytes = Arc::new(AtomicU64::new(0));
    let stream_error = Arc::new(Mutex::new(None));
    let payload_stream_error = stream_error.clone();
    let payload_per_file_bytes = per_file_bytes.clone();
    let stream = Box::pin(futures::stream::try_unfold(
        (
            field,
            payload_per_file_bytes,
            batch_bytes,
            payload_stream_error,
        ),
        |(mut field, per_file_bytes, batch_bytes, stream_error)| async move {
            let Some(chunk) = field.next().await else {
                return Ok(None);
            };
            let chunk = chunk.map_err(|error| anyhow::Error::msg(error.to_string()))?;
            if let Err(error) =
                consume_exercise_upload_bytes(&per_file_bytes, &batch_bytes, chunk.len())
            {
                *stream_error
                    .lock()
                    .unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(error);
                return Err(anyhow::Error::msg("Exercise upload exceeds the size limit"));
            }
            Ok(Some((
                chunk,
                (field, per_file_bytes, batch_bytes, stream_error),
            )))
        },
    ));
    (stream, stream_error, per_file_bytes)
}

fn validate_exercise_upload_file_count(files_received: usize) -> Result<(), ControllerError> {
    if files_received < EXERCISE_UPLOAD_MAX_FILES {
        return Ok(());
    }
    Err(controller_err!(
        BadRequest,
        format!("A maximum of {EXERCISE_UPLOAD_MAX_FILES} files can be uploaded at once")
    ))
}

fn validate_exercise_upload_id(
    field_name: &str,
    ids: &mut HashSet<String>,
) -> Result<(), ControllerError> {
    Uuid::parse_str(field_name).map_err(|_| {
        controller_err!(
            BadRequest,
            "Each exercise upload field name must be a UUID".to_string()
        )
    })?;
    if ids.insert(field_name.to_string()) {
        return Ok(());
    }
    Err(controller_err!(
        BadRequest,
        "Duplicate exercise upload field id".to_string()
    ))
}

fn validate_exercise_upload_filename(filename: Option<&str>) -> Result<&str, ControllerError> {
    filename
        .filter(|filename| !filename.is_empty())
        .ok_or_else(|| {
            controller_err!(
                BadRequest,
                "Every exercise upload part must be a file with a filename".to_string()
            )
        })
}

fn validate_exercise_upload_not_empty(files_received: usize) -> Result<(), ControllerError> {
    if files_received > 0 {
        return Ok(());
    }
    Err(controller_err!(
        BadRequest,
        "At least one file must be uploaded".to_string()
    ))
}

fn consume_exercise_upload_bytes(
    per_file_bytes: &AtomicU64,
    batch_bytes: &AtomicU64,
    chunk_len: usize,
) -> Result<(), ControllerError> {
    let chunk_len = u64::try_from(chunk_len).map_err(|error| {
        controller_err!(
            BadRequest,
            "exercise upload chunk length overflow".to_string(),
            error
        )
    })?;
    let file_total = per_file_bytes.fetch_add(chunk_len, Ordering::Relaxed) + chunk_len;
    let batch_total = batch_bytes.fetch_add(chunk_len, Ordering::Relaxed) + chunk_len;
    if file_total <= EXERCISE_UPLOAD_MAX_FILE_BYTES
        && batch_total <= EXERCISE_UPLOAD_MAX_BATCH_BYTES
    {
        return Ok(());
    }
    Err(controller_err!(
        BadRequest,
        "Exercise upload exceeds the 100 MiB per-file or batch limit"
    ))
}

#[cfg(test)]
mod exercise_upload_tests {
    use super::*;

    #[test]
    fn exercise_upload_ids_must_be_unique_uuids() {
        let id = Uuid::new_v4().to_string();
        let mut ids = HashSet::new();

        assert!(validate_exercise_upload_id(&id, &mut ids).is_ok());
        assert!(
            validate_exercise_upload_id(&id, &mut ids)
                .unwrap_err()
                .to_string()
                .contains("Duplicate")
        );
        assert!(
            validate_exercise_upload_id("filename.pdf", &mut ids)
                .unwrap_err()
                .to_string()
                .contains("must be a UUID")
        );
    }

    #[test]
    fn exercise_upload_limits_are_enforced_from_streamed_bytes() {
        let per_file = AtomicU64::new(EXERCISE_UPLOAD_MAX_FILE_BYTES - 1);
        let batch = AtomicU64::new(EXERCISE_UPLOAD_MAX_BATCH_BYTES - 1);
        assert!(consume_exercise_upload_bytes(&per_file, &batch, 1).is_ok());
        let error = consume_exercise_upload_bytes(&per_file, &batch, 1).unwrap_err();
        assert!(matches!(
            error.error_type(),
            ControllerErrorType::BadRequest
        ));

        let per_file = AtomicU64::new(0);
        let batch = AtomicU64::new(EXERCISE_UPLOAD_MAX_BATCH_BYTES);
        let error = consume_exercise_upload_bytes(&per_file, &batch, 1).unwrap_err();
        assert!(matches!(
            error.error_type(),
            ControllerErrorType::BadRequest
        ));
    }

    #[test]
    fn exercise_upload_rejects_an_eleventh_file() {
        assert!(validate_exercise_upload_file_count(EXERCISE_UPLOAD_MAX_FILES - 1).is_ok());
        assert!(validate_exercise_upload_file_count(EXERCISE_UPLOAD_MAX_FILES).is_err());
    }

    #[test]
    fn exercise_upload_rejects_empty_and_non_file_parts() {
        assert!(validate_exercise_upload_not_empty(0).is_err());
        assert!(validate_exercise_upload_not_empty(1).is_ok());
        assert!(validate_exercise_upload_filename(None).is_err());
        assert!(validate_exercise_upload_filename(Some("")).is_err());
        assert_eq!(
            validate_exercise_upload_filename(Some("report.pdf")).unwrap(),
            "report.pdf"
        );
    }
}

#[derive(Debug, Clone, Copy, Deserialize)]

pub enum StoreKind {
    Organization(Uuid),
    Course(Uuid),
    Exam(Uuid),
}

/// Processes an upload from CMS.
pub async fn upload_file_from_cms(
    headers: &HeaderMap,
    mut payload: Multipart,
    store_kind: StoreKind,
    file_store: &dyn FileStore,
    conn: &mut PgConnection,
    user: AuthUser,
) -> Result<PathBuf, ControllerError> {
    let file_payload = payload.next().await.ok_or_else(|| {
        ControllerError::new(ControllerErrorType::BadRequest, "Missing form data", None)
    })?;
    match file_payload {
        Ok(field) => {
            upload_field_from_cms(headers, field, store_kind, file_store, conn, user).await
        }
        Err(err) => Err(ControllerError::new(
            ControllerErrorType::InternalServerError,
            err.to_string(),
            None,
        )),
    }
}

/// Processes an upload from CMS.
pub async fn upload_field_from_cms(
    headers: &HeaderMap,
    field: Field,
    store_kind: StoreKind,
    file_store: &dyn FileStore,
    conn: &mut PgConnection,
    user: AuthUser,
) -> Result<PathBuf, ControllerError> {
    validate_media_headers(headers, &user, conn).await?;
    let path = match field.content_type().map(|ct| ct.type_()) {
        Some(mime::AUDIO) => generate_audio_path(&field, store_kind)?,
        Some(mime::IMAGE) => generate_image_path(&field, store_kind)?,
        _ => generate_file_path(&field, store_kind)?,
    };
    upload_field_to_storage(conn, &path, field, file_store, Some(user)).await?;
    Ok(path)
}

/// Processes an upload for an organization's image.
pub async fn upload_image_for_organization(
    headers: &HeaderMap,
    mut payload: Multipart,
    organization: &DatabaseOrganization,
    file_store: &Arc<dyn FileStore>,
    user: AuthUser,
    conn: &mut PgConnection,
) -> Result<PathBuf, ControllerError> {
    validate_media_headers(headers, &user, conn).await?;
    let next_payload: Result<Field, mp::MultipartError> =
        payload.next().await.ok_or_else(|| {
            ControllerError::new(ControllerErrorType::BadRequest, "Missing form data", None)
        })?;
    match next_payload {
        Ok(field) => {
            let path: PathBuf = match field.content_type().map(|ct| ct.type_()) {
                Some(mime::IMAGE) => {
                    generate_image_path(&field, StoreKind::Organization(organization.id))
                }
                Some(unsupported) => Err(ControllerError::new(
                    ControllerErrorType::BadRequest,
                    format!("Unsupported image Mime type: {}", unsupported),
                    None,
                )),
                None => Err(ControllerError::new(
                    ControllerErrorType::BadRequest,
                    "Missing image Mime type",
                    None,
                )),
            }?;
            upload_field_to_storage(conn, &path, field, file_store.as_ref(), Some(user)).await?;
            Ok(path)
        }
        Err(err) => Err(ControllerError::new(
            ControllerErrorType::InternalServerError,
            err.to_string(),
            None,
        )),
    }
}

// These limits must match the limits in CMS/src/services/backend/media/uploadMediaToServer.ts
// If you modify these, update the TypeScript file as well.
// Note: The nginx ingress also has a limit on max request size (see kubernetes/base/ingress.yml)
const FILE_SIZE_LIMITS: &[(mime::Name, i32)] = &[
    // 10 MB for images
    (mime::IMAGE, 10 * 1024 * 1024),
    // 100 MB for audio
    (mime::AUDIO, 100 * 1024 * 1024),
    // 100 MB for video
    (mime::VIDEO, 100 * 1024 * 1024),
    // 25 MB for documents/other files
    (mime::APPLICATION, 25 * 1024 * 1024),
];
// 10 MB default fallback
const DEFAULT_FILE_SIZE_LIMIT: i32 = 10 * 1024 * 1024;

fn get_size_limit_for_mime(mime_type: Option<mime::Name>) -> i32 {
    mime_type
        .and_then(|mime| FILE_SIZE_LIMITS.iter().find(|(m, _)| *m == mime))
        .map(|(_, size)| *size)
        .unwrap_or(DEFAULT_FILE_SIZE_LIMIT)
}

/// Uploads the data from the multipart `field` to the given `path` in file storage.
async fn upload_field_to_storage(
    conn: &mut PgConnection,
    path: &Path,
    field: mp::Field,
    file_store: &dyn FileStore,
    uploader: Option<AuthUser>,
) -> Result<(), ControllerError> {
    // Check file size limit based on mime type
    let mime_type = field.content_type().map(|ct| ct.type_());
    let size_limit = get_size_limit_for_mime(mime_type);

    // Get size from content disposition if available
    // Note: This does not enforce the size of the file since the client can lie about the content length
    if let Some(content_disposition) = field.content_disposition()
        && let Some(size_str) = content_disposition
            .parameters
            .iter()
            .find_map(|p| p.as_unknown("size"))
        && let Ok(size) = size_str.parse::<u64>()
        && size > size_limit as u64
    {
        return Err(ControllerError::new(
            ControllerErrorType::BadRequest,
            format!(
                "File size {} exceeds limit of {} bytes for type {}",
                size,
                size_limit,
                mime_type.map_or("unknown".to_string(), |m| m.to_string())
            ),
            None,
        ));
    }

    // TODO: convert archives into a uniform format
    let mime_type = field
        .content_type()
        .map(|ct| ct.to_string())
        .unwrap_or_default();

    let name = {
        let name_ref = field.name().ok_or_else(|| {
            ControllerError::new(
                ControllerErrorType::BadRequest,
                "Tried to upload a file without a file name".to_string(),
                None,
            )
        })?;
        name_ref.to_string()
    };

    let contents = Box::pin(field.map_err(|orig| anyhow::Error::msg(orig.to_string())));

    upload_file_to_storage(
        conn,
        path,
        &name,
        &mime_type,
        contents,
        file_store,
        uploader.map(|u| u.id),
    )
    .await?;
    Ok(())
}
pub async fn upload_certificate_svg(
    conn: &mut PgConnection,
    file_name: &str,
    file: GenericPayload,
    file_store: &dyn FileStore,
    course_id: Uuid,
    uploader: AuthUser,
) -> Result<(Uuid, PathBuf), ControllerError> {
    let path = path(file_name, FileType::Image, StoreKind::Course(course_id));
    let safe_path = make_filename_safe(&path);
    let id = upload_file_to_storage(
        conn,
        &safe_path,
        file_name,
        "image/svg+xml",
        file,
        file_store,
        Some(uploader.id),
    )
    .await?;
    Ok((id, safe_path))
}

async fn upload_file_to_storage(
    conn: &mut PgConnection,
    path: &Path,
    file_name: &str,
    mime_type: &str,
    file: GenericPayload,
    file_store: &dyn FileStore,
    uploader: Option<Uuid>,
) -> Result<Uuid, ControllerError> {
    let mut tx = conn.begin().await?;
    let id = upload_file_to_storage_in_existing_transaction(
        &mut tx, path, file_name, mime_type, file, file_store, uploader,
    )
    .await?;
    tx.commit().await?;
    Ok(id)
}

async fn upload_file_to_storage_in_existing_transaction(
    conn: &mut PgConnection,
    path: &Path,
    file_name: &str,
    mime_type: &str,
    file: GenericPayload,
    file_store: &dyn FileStore,
    uploader: Option<Uuid>,
) -> Result<Uuid, ControllerError> {
    let path_string = path.to_str().context("invalid path")?.to_string();
    let id = models::file_uploads::insert(conn, file_name, &path_string, mime_type, uploader, None)
        .await?;
    file_store.upload_stream(path, file, mime_type).await?;
    Ok(id)
}

fn make_filename_safe(path: &PathBuf) -> PathBuf {
    let mut path_buf = path.to_owned();
    let random_string = Alphanumeric.sample_string(&mut rand::rng(), 25);
    path_buf.set_file_name(random_string);
    if let Some(ext) = path.extension() {
        // For convenience, we'll keep the original extension in most cases. We'll just filter out any potentially problematic characters.
        let ext = ext
            .to_str()
            .unwrap_or("")
            .chars()
            .filter(|c| c.is_alphanumeric())
            .collect::<String>();
        path_buf.set_extension(ext);
    }
    path_buf
}

pub async fn delete_file_from_storage(
    conn: &mut PgConnection,
    id: Uuid,
    file_store: &dyn FileStore,
) -> Result<(), ControllerError> {
    let file_to_delete = models::file_uploads::delete_and_fetch_path(conn, id).await?;
    file_store.delete(Path::new(&file_to_delete)).await?;
    Ok(())
}

/// Generates a path for an audio file with the appropriate extension.
fn generate_audio_path(field: &Field, store_kind: StoreKind) -> Result<PathBuf, ControllerError> {
    let extension = match field
        .content_type()
        .map(|ct| ct.to_string())
        .unwrap_or_default()
        .as_str()
    {
        "audio/aac" => ".aac",
        "audio/mpeg" => ".mp3",
        "audio/ogg" => ".oga",
        "audio/opus" => ".opus",
        "audio/wav" => ".wav",
        "audio/webm" => ".weba",
        "audio/midi" => ".mid",
        "audio/x-midi" => ".mid",
        unsupported => {
            return Err(ControllerError::new(
                ControllerErrorType::BadRequest,
                format!("Unsupported audio Mime type: {}", unsupported),
                None,
            ));
        }
    };
    let mut file_name = generate_random_string(30);
    file_name.push_str(extension);
    let path = path(&file_name, FileType::Audio, store_kind);
    Ok(path)
}

/// Generates a path for a generic file with the appropriate extension based on its filename.
fn generate_file_path(field: &Field, store_kind: StoreKind) -> Result<PathBuf, ControllerError> {
    let field_content = field.content_disposition().ok_or_else(|| {
        ControllerError::new(
            ControllerErrorType::BadRequest,
            "No content disposition in uploaded file".to_string(),
            None,
        )
    })?;
    let field_content_name = field_content.get_filename().ok_or_else(|| {
        ControllerError::new(
            ControllerErrorType::BadRequest,
            "Missing file name in content-disposition",
            None,
        )
    })?;

    let mut file_name = generate_random_string(30);
    let uploaded_file_extension = get_extension_from_filename(field_content_name);
    if let Some(extension) = uploaded_file_extension {
        file_name.push_str(format!(".{}", extension).as_str());
    }

    let path = path(&file_name, FileType::File, store_kind);
    Ok(path)
}

/// Generates a path for an image file with the appropriate extension.
fn generate_image_path(field: &Field, store_kind: StoreKind) -> Result<PathBuf, ControllerError> {
    let extension = match field
        .content_type()
        .map(|ct| ct.to_string())
        .unwrap_or_default()
        .as_str()
    {
        "image/jpeg" => ".jpg",
        "image/png" => ".png",
        "image/svg+xml" => ".svg",
        "image/tiff" => ".tif",
        "image/bmp" => ".bmp",
        "image/webp" => ".webp",
        "image/gif" => ".gif",
        unsupported => {
            return Err(ControllerError::new(
                ControllerErrorType::BadRequest,
                format!("Unsupported image Mime type: {}", unsupported),
                None,
            ));
        }
    };

    // using a random string for the image name because
    // a) we don't want the filename to be user controllable
    // b) we don't want the filename to be too easily guessable (so no uuid)
    let mut file_name = generate_random_string(30);
    file_name.push_str(extension);
    let path = path(&file_name, FileType::Image, store_kind);
    Ok(path)
}

/// Generates a path for an audio file with the appropriate extension.
async fn validate_media_headers(
    headers: &HeaderMap,
    user: &AuthUser,
    conn: &mut PgConnection,
) -> ControllerResult<()> {
    let content_type = headers.get(header::CONTENT_TYPE).ok_or_else(|| {
        ControllerError::new(
            ControllerErrorType::BadRequest,
            "Please provide a Content-Type header",
            None,
        )
    })?;
    let content_type_string = String::from_utf8_lossy(content_type.as_bytes()).to_string();

    if !content_type_string.contains("multipart/form-data") {
        return Err(ControllerError::new(
            ControllerErrorType::BadRequest,
            format!("Unsupported type: {}", content_type_string),
            None,
        ));
    }

    let content_length = headers.get(header::CONTENT_LENGTH).ok_or_else(|| {
        ControllerError::new(
            ControllerErrorType::BadRequest,
            "Please provide a Content-Length in header",
            None,
        )
    })?;
    let content_length_number = String::from_utf8_lossy(content_length.as_bytes())
        .to_string()
        .parse::<i32>()
        .map_err(|original_err| {
            ControllerError::new(
                ControllerErrorType::InternalServerError,
                original_err.to_string(),
                Some(original_err.into()),
            )
        })?;

    let mime_type = headers
        .get("X-File-Type")
        .map(|h| h.to_str().unwrap_or("application/octet-stream"))
        .unwrap_or("application/octet-stream")
        .split('/')
        .next()
        .map(|s| match s {
            "image" => mime::IMAGE,
            "audio" => mime::AUDIO,
            "video" => mime::VIDEO,
            "application" => mime::APPLICATION,
            _ => mime::APPLICATION,
        });
    let size_limit = get_size_limit_for_mime(mime_type);

    // Note: This does not enforce the size of the file since the client can lie about the content length
    if content_length_number > size_limit {
        return Err(ControllerError::new(
            ControllerErrorType::BadRequest,
            format!(
                "File size {} exceeds limit of {} bytes for type {}",
                content_length_number,
                size_limit,
                mime_type.map_or("unknown".to_string(), |m| m.to_string())
            ),
            None,
        ));
    }

    let token = authorize(conn, Act::Teach, Some(user.id), Res::AnyCourse).await?;
    token.authorized_ok(())
}

enum FileType {
    Image,
    Audio,
    File,
}

fn path(file_name: &str, file_type: FileType, store_kind: StoreKind) -> PathBuf {
    let (base_dir, base_id) = match store_kind {
        StoreKind::Organization(id) => ("organization", id),
        StoreKind::Course(id) => ("course", id),
        StoreKind::Exam(id) => ("exam", id),
    };
    let file_type_subdir = match file_type {
        FileType::Image => "images",
        FileType::Audio => "audios",
        FileType::File => "files",
    };
    [base_dir, &base_id.to_string(), file_type_subdir, file_name]
        .iter()
        .collect()
}