cadmus 0.1.4

//! File uploads: one-shot multipart uploads, resumable chunked uploads, and downloads, all
//! streamed through [`Storage`].
//!
//! Resumable protocol (any chunk size up to the configured maximum, in order):
//!
//! 1. `POST   {api}/uploads` with `{"filename", "size", "content_type"}` creates a session.
//! 2. `PATCH  {api}/uploads/{id}` with header `Upload-Offset: N` and the raw bytes appends a
//!    chunk. The offset must equal what the server has; otherwise it answers 409 with the
//!    offset it expects.
//! 3. `GET    {api}/uploads/{id}` reports the current offset, which is how a client resumes
//!    after a dropped connection or a restart on either side.
//! 4. The chunk that reaches `size` completes the upload and returns the stored file.
//!
//! Session state lives in the database and chunks in the storage backend, so an upload
//! survives server restarts and can continue on a different instance.

use std::sync::Arc;

use axum::body::{Body, Bytes};
use axum::extract::{DefaultBodyLimit, Multipart, Path, Query, State};
use axum::handler::Handler;
use axum::http::header::{CONTENT_DISPOSITION, CONTENT_LENGTH, CONTENT_TYPE, LOCATION};
use axum::http::{HeaderMap, HeaderValue, StatusCode};
use axum::response::{IntoResponse, Response};
use axum::routing::{get, post};
use axum::{Json, Router};
use sea_orm::schema::SchemaBuilder;
use sea_orm::sea_query::Expr;
use sea_orm::{
    ActiveModelTrait, ColumnTrait, DatabaseConnection, EntityTrait, PaginatorTrait, QueryFilter,
    QueryOrder, Set,
};
use serde::Deserialize;
use serde_json::json;

use crate::AdminState;
use crate::access::Action;
use crate::api::bad_request;
use crate::auth::Session;
use crate::error::{AdminError, ApiError, ApiResult};
use crate::storage::Storage;

const OFFSET_HEADER: &str = "upload-offset";
pub(crate) const DEFAULT_MAX_FILE_BYTES: u64 = 1024 * 1024 * 1024;
pub(crate) const DEFAULT_MAX_CHUNK_BYTES: usize = 8 * 1024 * 1024;
/// Room for multipart boundaries and part headers on top of the file itself.
const MULTIPART_OVERHEAD_BYTES: usize = 64 * 1024;
const FILES_PER_PAGE: u64 = 25;

/// The `storage_file` entity: one stored file.
pub mod file {
    use sea_orm::entity::prelude::*;

    /// Metadata for a stored file. The bytes live in the storage backend under `key`.
    #[derive(Clone, Debug, PartialEq, Eq, DeriveEntityModel)]
    #[sea_orm(table_name = "storage_file")]
    pub struct Model {
        /// Primary key.
        #[sea_orm(primary_key)]
        pub id: i32,
        /// Storage key, generated by the server.
        #[sea_orm(unique)]
        pub key: String,
        /// Name supplied by the uploader, for display and downloads only.
        pub filename: String,
        /// Media type supplied by the uploader (not verified).
        pub content_type: String,
        /// Size in bytes.
        pub size: i64,
        /// Account that uploaded it.
        pub created_by: String,
    }

    /// No relations.
    #[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]
    pub enum Relation {}

    impl ActiveModelBehavior for ActiveModel {}
}

/// The `storage_upload` entity: one resumable upload in progress.
pub mod session {
    use sea_orm::entity::prelude::*;

    /// State of a resumable upload.
    #[derive(Clone, Debug, PartialEq, Eq, DeriveEntityModel)]
    #[sea_orm(table_name = "storage_upload")]
    pub struct Model {
        /// Random session id.
        #[sea_orm(primary_key, auto_increment = false)]
        pub id: String,
        /// Storage key the finished file will have.
        pub key: String,
        /// Name supplied by the uploader.
        pub filename: String,
        /// Media type supplied by the uploader.
        pub content_type: String,
        /// Declared total size in bytes.
        pub size: i64,
        /// Bytes received so far; the offset the next chunk must start at.
        pub received: i64,
        /// Comma-separated start offsets of the stored chunks, in order.
        #[sea_orm(column_type = "Text")]
        pub parts: String,
        /// Account that started the upload.
        pub created_by: String,
    }

    /// No relations.
    #[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]
    pub enum Relation {}

    impl ActiveModelBehavior for ActiveModel {}
}

/// Result of appending a chunk.
#[derive(Debug)]
pub enum Progress {
    /// More bytes are expected, starting at this offset.
    Incomplete {
        /// Offset the next chunk must start at.
        received: u64,
    },
    /// The upload is finished and stored.
    Complete(file::Model),
}

/// The upload service: storage backend, database and limits.
#[derive(Clone)]
pub struct Uploads {
    db: DatabaseConnection,
    storage: Storage,
    max_file_bytes: u64,
    max_chunk_bytes: usize,
}

impl Uploads {
    /// Creates the service with default limits (1 GiB per file, 8 MiB per chunk).
    #[must_use]
    pub fn new(db: DatabaseConnection, storage: Storage) -> Self {
        Self {
            db,
            storage,
            max_file_bytes: DEFAULT_MAX_FILE_BYTES,
            max_chunk_bytes: DEFAULT_MAX_CHUNK_BYTES,
        }
    }

    /// Reads `STORAGE_URL` (see [`Storage::from_env`]), `UPLOAD_MAX_FILE_BYTES` and
    /// `UPLOAD_MAX_CHUNK_BYTES`. Returns `None` when `STORAGE_URL` is unset.
    ///
    /// # Errors
    /// Returns [`AdminError::Config`] if a variable is invalid.
    pub fn from_env(db: DatabaseConnection) -> Result<Option<Self>, AdminError> {
        let Some(storage) = Storage::from_env()? else {
            return Ok(None);
        };
        let mut uploads = Self::new(db, storage);
        let positive = |key: &str| -> Result<Option<u64>, AdminError> {
            match std::env::var(key) {
                Ok(text) => text
                    .parse::<u64>()
                    .ok()
                    .filter(|n| *n > 0)
                    .map(Some)
                    .ok_or_else(|| AdminError::Config(format!("{key} must be a positive integer"))),
                Err(_) => Ok(None),
            }
        };
        if let Some(bytes) = positive("UPLOAD_MAX_FILE_BYTES")? {
            uploads.max_file_bytes = bytes;
        }
        if let Some(bytes) = positive("UPLOAD_MAX_CHUNK_BYTES")? {
            uploads.max_chunk_bytes = usize::try_from(bytes).unwrap_or(usize::MAX);
        }
        Ok(Some(uploads))
    }

    /// Largest file accepted, in bytes.
    #[must_use]
    pub fn max_file_bytes(mut self, bytes: u64) -> Self {
        self.max_file_bytes = bytes.max(1);
        self
    }

    /// Largest chunk accepted by a resumable upload, in bytes.
    #[must_use]
    pub fn max_chunk_bytes(mut self, bytes: usize) -> Self {
        self.max_chunk_bytes = bytes.max(1);
        self
    }

    /// Largest file accepted, in bytes.
    pub(crate) fn file_limit(&self) -> u64 {
        self.max_file_bytes
    }

    /// The storage backend.
    #[must_use]
    pub fn storage(&self) -> &Storage {
        &self.storage
    }

    pub(crate) fn register_schema(builder: SchemaBuilder) -> SchemaBuilder {
        builder.register(file::Entity).register(session::Entity)
    }

    async fn record(
        &self,
        key: String,
        filename: String,
        content_type: String,
        size: u64,
        owner: &str,
    ) -> Result<file::Model, AdminError> {
        let row = file::ActiveModel {
            key: Set(key.clone()),
            filename: Set(filename),
            content_type: Set(content_type),
            size: Set(to_i64(size)?),
            created_by: Set(owner.to_owned()),
            ..Default::default()
        };
        match row.insert(&self.db).await {
            Ok(saved) => Ok(saved),
            Err(error) => {
                // Do not leave bytes behind that no record points to.
                if let Err(cleanup) = self.storage.delete(&key).await {
                    tracing::warn!(error = %cleanup, "could not remove orphaned object");
                }
                Err(error.into())
            }
        }
    }

    /// Looks up a stored file's metadata.
    ///
    /// # Errors
    /// Returns [`AdminError::NotFound`] if there is no such file.
    pub async fn file(&self, id: i32) -> Result<file::Model, AdminError> {
        file::Entity::find_by_id(id)
            .one(&self.db)
            .await?
            .ok_or_else(|| AdminError::NotFound(format!("file `{id}`")))
    }

    /// Deletes a stored file and its record.
    ///
    /// # Errors
    /// Returns [`AdminError::NotFound`] if there is no such file.
    pub async fn remove(&self, id: i32) -> Result<(), AdminError> {
        let found = self.file(id).await?;
        // Record first: a dangling object is invisible, a dangling record is a broken download.
        file::Entity::delete_by_id(id).exec(&self.db).await?;
        self.storage.delete(&found.key).await
    }

    /// Starts a resumable upload.
    ///
    /// # Errors
    /// Returns [`AdminError::TooLarge`] if `size` exceeds the file limit.
    pub async fn begin(
        &self,
        filename: &str,
        content_type: &str,
        size: u64,
        owner: &str,
    ) -> Result<session::Model, AdminError> {
        if size > self.max_file_bytes {
            return Err(AdminError::TooLarge(format!(
                "files may be at most {} bytes",
                self.max_file_bytes
            )));
        }
        let filename = clean_filename(filename);
        let row = session::ActiveModel {
            id: Set(random_hex()),
            key: Set(new_key(&filename)),
            filename: Set(filename),
            content_type: Set(clean_content_type(content_type)),
            size: Set(to_i64(size)?),
            received: Set(0),
            parts: Set(String::new()),
            created_by: Set(owner.to_owned()),
        };
        Ok(row.insert(&self.db).await?)
    }

    /// Looks up an upload session.
    ///
    /// # Errors
    /// Returns [`AdminError::NotFound`] if there is no such session (including finished ones).
    pub async fn session(&self, id: &str) -> Result<session::Model, AdminError> {
        session::Entity::find_by_id(id.to_owned())
            .one(&self.db)
            .await?
            .ok_or_else(|| AdminError::NotFound("upload".into()))
    }

    /// Appends a chunk that starts at `offset`.
    ///
    /// An empty chunk at the final offset retries completion, for the case where every byte
    /// arrived but storing the finished file failed.
    ///
    /// # Errors
    /// - [`AdminError::Conflict`] if `offset` is not the next expected offset (the message
    ///   carries the expected one), or another request advanced the upload concurrently.
    /// - [`AdminError::TooLarge`] if the chunk exceeds the chunk limit or runs past the
    ///   declared size.
    pub async fn append(
        &self,
        id: &str,
        offset: u64,
        chunk: Bytes,
    ) -> Result<Progress, AdminError> {
        let current = self.session(id).await?;
        let received = to_u64(current.received);
        let size = to_u64(current.size);
        if offset != received {
            return Err(AdminError::Conflict(format!("expected offset {received}")));
        }
        let length = chunk.len() as u64;
        if chunk.len() > self.max_chunk_bytes {
            return Err(AdminError::TooLarge(format!(
                "chunks may be at most {} bytes",
                self.max_chunk_bytes
            )));
        }
        if received + length > size {
            return Err(AdminError::TooLarge(
                "the chunk runs past the declared size".into(),
            ));
        }
        if length > 0 {
            self.storage.put(&part_key(id, offset), chunk).await?;
            let parts = if current.parts.is_empty() {
                offset.to_string()
            } else {
                format!("{},{offset}", current.parts)
            };
            // Compare-and-set on the offset: of two racing requests, exactly one advances.
            let updated = session::Entity::update_many()
                .col_expr(
                    session::Column::Received,
                    Expr::value(to_i64(received + length)?),
                )
                .col_expr(session::Column::Parts, Expr::value(parts))
                .filter(session::Column::Id.eq(id))
                .filter(session::Column::Received.eq(current.received))
                .exec(&self.db)
                .await?;
            if updated.rows_affected == 0 {
                return Err(AdminError::Conflict(
                    "another request advanced this upload; query its offset and retry".into(),
                ));
            }
        }
        if received + length < size {
            return Ok(Progress::Incomplete {
                received: received + length,
            });
        }
        self.complete(id).await.map(Progress::Complete)
    }

    /// Assembles the chunks into the final object and records the file.
    async fn complete(&self, id: &str) -> Result<file::Model, AdminError> {
        let done = self.session(id).await?;
        let offsets: Vec<u64> = done
            .parts
            .split(',')
            .filter_map(|text| text.parse().ok())
            .collect();
        let mut sink = self.storage.sink(&done.key);
        for offset in &offsets {
            let bytes = match self.storage.get(&part_key(id, *offset)).await {
                Ok(part) => part.bytes().await.map_err(AdminError::from),
                Err(error) => Err(error),
            };
            let written = match bytes {
                Ok(bytes) => sink.write(&bytes).await,
                Err(error) => Err(error),
            };
            if let Err(error) = written {
                sink.abort().await;
                return Err(error);
            }
        }
        let size = sink.finish().await?;
        let saved = self
            .record(
                done.key.clone(),
                done.filename.clone(),
                done.content_type.clone(),
                size,
                &done.created_by,
            )
            .await
            .map_err(|error| match error {
                // Two completions raced; the other one owns the file.
                AdminError::Db(db) if db.sql_err().is_some() => {
                    AdminError::Conflict("this upload was already completed".into())
                }
                other => other,
            })?;
        self.discard(&done.id, &offsets).await?;
        Ok(saved)
    }

    async fn discard(&self, id: &str, offsets: &[u64]) -> Result<(), AdminError> {
        session::Entity::delete_by_id(id.to_owned())
            .exec(&self.db)
            .await?;
        for offset in offsets {
            if let Err(error) = self.storage.delete(&part_key(id, *offset)).await {
                tracing::warn!(%error, "could not remove upload chunk");
            }
        }
        Ok(())
    }

    /// Cancels a resumable upload and deletes its chunks.
    ///
    /// # Errors
    /// Returns [`AdminError::NotFound`] if there is no such session.
    pub async fn abort(&self, id: &str) -> Result<(), AdminError> {
        let current = self.session(id).await?;
        let offsets: Vec<u64> = current
            .parts
            .split(',')
            .filter_map(|text| text.parse().ok())
            .collect();
        self.discard(id, &offsets).await
    }
}

fn to_i64(value: u64) -> Result<i64, AdminError> {
    i64::try_from(value).map_err(|_| AdminError::TooLarge("size does not fit the database".into()))
}

fn to_u64(value: i64) -> u64 {
    u64::try_from(value).unwrap_or(0)
}

fn random_hex() -> String {
    rand::random::<[u8; 16]>()
        .iter()
        .map(|b| format!("{b:02x}"))
        .collect()
}

fn part_key(id: &str, offset: u64) -> String {
    // Zero-padded so chunk objects list in order when inspected by hand.
    format!("uploads/{id}/{offset:020}")
}

/// Storage keys never contain anything the client sent except a vetted extension, so a
/// hostile filename cannot choose where bytes land.
fn new_key(filename: &str) -> String {
    let id = random_hex();
    let extension = filename
        .rsplit_once('.')
        .map(|(_, ext)| ext.to_ascii_lowercase())
        .filter(|ext| {
            !ext.is_empty() && ext.len() <= 10 && ext.chars().all(|c| c.is_ascii_alphanumeric())
        });
    match extension {
        Some(ext) => format!("files/{}/{id}.{ext}", &id[..2]),
        None => format!("files/{}/{id}", &id[..2]),
    }
}

fn clean_filename(raw: &str) -> String {
    let base = raw.rsplit(['/', '\\']).next().unwrap_or("");
    let cleaned: String = base.chars().filter(|c| !c.is_control()).take(255).collect();
    let cleaned = cleaned.trim().to_owned();
    if cleaned.is_empty() || cleaned == "." || cleaned == ".." {
        "upload".to_owned()
    } else {
        cleaned
    }
}

fn clean_content_type(raw: &str) -> String {
    let valid = !raw.is_empty()
        && raw.len() <= 127
        && raw.contains('/')
        && raw
            .chars()
            .all(|c| c.is_ascii_alphanumeric() || "/.+-_;= ".contains(c));
    if valid {
        raw.to_owned()
    } else {
        "application/octet-stream".to_owned()
    }
}

// ---- HTTP ---------------------------------------------------------------------------------

type Shared = State<Arc<AdminState>>;

/// The name file permissions are granted under, as if uploads were a model.
pub(crate) const FILES: &str = "files";

pub(crate) fn router(state: Arc<AdminState>, uploads: &Uploads) -> Router {
    let b = state.api_base.clone();
    let whole_file = usize::try_from(uploads.max_file_bytes)
        .unwrap_or(usize::MAX)
        .saturating_add(MULTIPART_OVERHEAD_BYTES);
    Router::new()
        .route(
            &format!("{b}/files"),
            get(list_files).post(upload_multipart.layer(DefaultBodyLimit::max(whole_file))),
        )
        .route(
            &format!("{b}/files/{{id}}"),
            get(file_info).delete(delete_file),
        )
        .route(&format!("{b}/files/{{id}}/content"), get(download))
        .route(&format!("{b}/uploads"), post(begin_upload))
        .route(
            &format!("{b}/uploads/{{id}}"),
            get(upload_status)
                .patch(append_chunk.layer(DefaultBodyLimit::max(uploads.max_chunk_bytes)))
                .delete(abort_upload),
        )
        .with_state(state)
}

fn service(state: &AdminState) -> Result<&Uploads, AdminError> {
    state
        .uploads
        .as_ref()
        .ok_or_else(|| AdminError::NotFound("file storage is not configured".into()))
}

fn file_json(state: &AdminState, file: &file::Model) -> serde_json::Value {
    json!({
        "id": file.id,
        "filename": file.filename,
        "content_type": file.content_type,
        "size": file.size,
        "created_by": file.created_by,
        "url": format!("{}/files/{}/content", state.api_base, file.id),
    })
}

fn created(state: &AdminState, file: &file::Model) -> Response {
    let mut response = (StatusCode::CREATED, Json(file_json(state, file))).into_response();
    if let Ok(location) = HeaderValue::from_str(&format!("{}/files/{}", state.api_base, file.id)) {
        response.headers_mut().insert(LOCATION, location);
    }
    response
}

impl Uploads {
    /// Streams one part of a multipart body into storage and records the file.
    pub(crate) async fn store_field(
        &self,
        mut field: axum::extract::multipart::Field<'_>,
        owner: &str,
    ) -> Result<file::Model, AdminError> {
        let filename = clean_filename(field.file_name().unwrap_or(""));
        let content_type = clean_content_type(field.content_type().unwrap_or(""));
        let key = new_key(&filename);
        let mut sink = self.storage.sink(&key);
        loop {
            let step = match field.chunk().await {
                Ok(Some(bytes)) if sink.written() + bytes.len() as u64 > self.max_file_bytes => {
                    Err(AdminError::TooLarge(format!(
                        "files may be at most {} bytes",
                        self.max_file_bytes
                    )))
                }
                Ok(Some(bytes)) => sink.write(&bytes).await,
                Ok(None) => break,
                Err(error) => Err(multipart_admin_error(&error)),
            };
            if let Err(error) = step {
                sink.abort().await;
                return Err(error);
            }
        }
        let size = sink.finish().await?;
        self.record(key, filename, content_type, size, owner).await
    }

    /// The files with these ids, by id; ids that name no file are left out.
    pub(crate) async fn files_by_id(
        &self,
        ids: &[i32],
    ) -> Result<std::collections::HashMap<i32, file::Model>, AdminError> {
        if ids.is_empty() {
            return Ok(std::collections::HashMap::new());
        }
        Ok(file::Entity::find()
            .filter(file::Column::Id.is_in(ids.iter().copied()))
            .all(&self.db)
            .await?
            .into_iter()
            .map(|f| (f.id, f))
            .collect())
    }

    /// The file's content as a response: always a download, except raster images when
    /// `inline` is asked for, which may be shown in an `<img>`.
    pub(crate) async fn respond(
        &self,
        found: &file::Model,
        inline: bool,
    ) -> Result<Response, AdminError> {
        let object = self.storage.get(&found.key).await?;
        let mut response = Body::from_stream(object.into_stream()).into_response();
        let headers = response.headers_mut();
        // The media type came from the uploader: serve it as a download and forbid sniffing,
        // so a hostile HTML or script file can never run in this origin. Only image types
        // browsers cannot execute (not SVG) may be shown inline.
        let shown = inline && is_raster_image(&found.content_type);
        let safe_name: String = found
            .filename
            .chars()
            .map(|c| {
                let safe = (c.is_ascii_graphic() && !matches!(c, '"' | '\\' | ';')) || c == ' ';
                if safe { c } else { '_' }
            })
            .collect();
        let pairs = [
            (CONTENT_TYPE, found.content_type.clone()),
            (CONTENT_LENGTH, found.size.to_string()),
            (
                CONTENT_DISPOSITION,
                format!(
                    "{}; filename=\"{safe_name}\"",
                    if shown { "inline" } else { "attachment" }
                ),
            ),
        ];
        for (name, value) in pairs {
            if let Ok(value) = HeaderValue::from_str(&value) {
                headers.insert(name, value);
            }
        }
        headers.insert(
            "x-content-type-options",
            HeaderValue::from_static("nosniff"),
        );
        Ok(response)
    }
}

/// Image types a browser only displays, never runs.
pub(crate) fn is_raster_image(content_type: &str) -> bool {
    matches!(
        content_type,
        "image/png" | "image/jpeg" | "image/gif" | "image/webp" | "image/avif"
    )
}

fn multipart_admin_error(error: &axum::extract::multipart::MultipartError) -> AdminError {
    if error.status() == StatusCode::PAYLOAD_TOO_LARGE {
        AdminError::TooLarge("the upload exceeds the size limit".into())
    } else {
        AdminError::BadInput {
            field: "file".into(),
            message: "the upload was cut off or malformed".into(),
        }
    }
}

async fn upload_multipart(
    State(state): Shared,
    session: Session,
    mut multipart: Multipart,
) -> ApiResult<Response> {
    session.require_bearer()?;
    session.require(FILES, Action::Add)?;
    let uploads = service(&state)?;
    loop {
        let field = match multipart.next_field().await {
            Ok(Some(field)) => field,
            Ok(None) => return Err(bad_request("send the file in a form field named `file`")),
            Err(error) => return Err(multipart_error(&error)),
        };
        if field.name() != Some("file") {
            continue;
        }
        let saved = uploads.store_field(field, &session.username).await?;
        return Ok(created(&state, &saved));
    }
}

fn multipart_error(error: &axum::extract::multipart::MultipartError) -> ApiError {
    if error.status() == StatusCode::PAYLOAD_TOO_LARGE {
        ApiError::from(AdminError::TooLarge(
            "the upload exceeds the size limit".into(),
        ))
    } else {
        bad_request("malformed multipart body")
    }
}

#[derive(Deserialize)]
struct PageParams {
    #[serde(default)]
    page: Option<u64>,
}

async fn list_files(
    State(state): Shared,
    session: Session,
    Query(params): Query<PageParams>,
) -> ApiResult<Json<serde_json::Value>> {
    session.require_bearer()?;
    session.require(FILES, Action::View)?;
    let uploads = service(&state)?;
    let page = params.page.unwrap_or(1).max(1);
    let paginator = file::Entity::find()
        .order_by_desc(file::Column::Id)
        .paginate(&uploads.db, FILES_PER_PAGE);
    let total = paginator.num_items().await.map_err(AdminError::from)?;
    let rows = paginator
        .fetch_page(page - 1)
        .await
        .map_err(AdminError::from)?;
    let data: Vec<_> = rows.iter().map(|f| file_json(&state, f)).collect();
    Ok(Json(json!({
        "data": data,
        "meta": { "total": total, "page": page, "per_page": FILES_PER_PAGE },
    })))
}

async fn file_info(
    State(state): Shared,
    session: Session,
    Path(id): Path<i32>,
) -> ApiResult<Json<serde_json::Value>> {
    session.require_bearer()?;
    session.require(FILES, Action::View)?;
    let found = service(&state)?.file(id).await?;
    Ok(Json(file_json(&state, &found)))
}

async fn download(
    State(state): Shared,
    session: Session,
    Path(id): Path<i32>,
) -> ApiResult<Response> {
    session.require_bearer()?;
    session.require(FILES, Action::View)?;
    let uploads = service(&state)?;
    let found = uploads.file(id).await?;
    Ok(uploads.respond(&found, false).await?)
}

async fn delete_file(
    State(state): Shared,
    session: Session,
    Path(id): Path<i32>,
) -> ApiResult<StatusCode> {
    session.require_bearer()?;
    session.require(FILES, Action::Delete)?;
    service(&state)?.remove(id).await?;
    Ok(StatusCode::NO_CONTENT)
}

#[derive(Deserialize)]
struct BeginUpload {
    filename: String,
    size: u64,
    #[serde(default)]
    content_type: Option<String>,
}

fn session_response(
    state: &AdminState,
    uploads: &Uploads,
    status: StatusCode,
    upload: &session::Model,
) -> Response {
    let body = Json(json!({
        "id": upload.id,
        "filename": upload.filename,
        "size": upload.size,
        "received": upload.received,
        "max_chunk_bytes": uploads.max_chunk_bytes,
        "complete": false,
        "url": format!("{}/uploads/{}", state.api_base, upload.id),
    }));
    let mut response = (status, body).into_response();
    if let Ok(offset) = HeaderValue::from_str(&upload.received.to_string()) {
        response.headers_mut().insert(OFFSET_HEADER, offset);
    }
    response
}

async fn begin_upload(State(state): Shared, session: Session, body: Bytes) -> ApiResult<Response> {
    session.require_bearer()?;
    session.require(FILES, Action::Add)?;
    let uploads = service(&state)?;
    let request: BeginUpload = serde_json::from_slice(&body)
        .map_err(|_| bad_request("expected JSON with `filename` and `size`"))?;
    let started = uploads
        .begin(
            &request.filename,
            request.content_type.as_deref().unwrap_or(""),
            request.size,
            &session.username,
        )
        .await?;
    Ok(session_response(
        &state,
        uploads,
        StatusCode::CREATED,
        &started,
    ))
}

async fn upload_status(
    State(state): Shared,
    session: Session,
    Path(id): Path<String>,
) -> ApiResult<Response> {
    session.require_bearer()?;
    session.require(FILES, Action::Add)?;
    let uploads = service(&state)?;
    let current = uploads.session(&id).await?;
    Ok(session_response(&state, uploads, StatusCode::OK, &current))
}

async fn append_chunk(
    State(state): Shared,
    session: Session,
    Path(id): Path<String>,
    headers: HeaderMap,
    body: Bytes,
) -> ApiResult<Response> {
    session.require_bearer()?;
    session.require(FILES, Action::Add)?;
    let uploads = service(&state)?;
    let offset = headers
        .get(OFFSET_HEADER)
        .and_then(|v| v.to_str().ok())
        .and_then(|v| v.trim().parse::<u64>().ok())
        .ok_or_else(|| bad_request("the Upload-Offset header is required"))?;
    match uploads.append(&id, offset, body).await? {
        Progress::Incomplete { .. } => {
            let current = uploads.session(&id).await?;
            Ok(session_response(&state, uploads, StatusCode::OK, &current))
        }
        Progress::Complete(saved) => Ok(created(&state, &saved)),
    }
}

async fn abort_upload(
    State(state): Shared,
    session: Session,
    Path(id): Path<String>,
) -> ApiResult<StatusCode> {
    session.require_bearer()?;
    session.require(FILES, Action::Add)?;
    service(&state)?.abort(&id).await?;
    Ok(StatusCode::NO_CONTENT)
}