//! File operations API.
//!
//! All operations live on the same URL shape as the listing, distinguished by
//! method (and content type for POST):
//!
//! - `GET    /api/files/{root_id}` and `/api/files/{root_id}/{*path}` — list
//! - `DELETE /api/files/{root_id}/{*path}` — delete a file or folder
//! - `POST   /api/files/{root_id}/{*path}` — create a folder (no body)
//! - `POST   /api/files/{root_id}/{*path}` (JSON body) — rename / move / copy
//! - `POST   /api/files/{root_id}/{*path}` (multipart) — upload into the dir

use std::io::{self, Read};
use std::path::{Component, PathBuf};
use std::sync::Arc;

use axum::Json;
use axum::extract::{Path as AxumPath, Query as AxumQuery, State};
use axum::http::{StatusCode, header};
use axum::response::{IntoResponse, Response};
use futures_util::StreamExt;
use multer::Multipart;
use serde::Deserialize;
use tokio::io::AsyncWriteExt;
use tokio::sync::mpsc;
use tokio_stream::wrappers::ReceiverStream;

use crate::api::common::{AuthUser, blocking, target_rel};
use crate::archive::{self, ArchiveFormat};
use crate::db::{RootRow, ShareRow};
use crate::error::{ApiError, AppState};
use crate::fs::{self, FsError};
use api_types::{
    DeleteResp, FilesResp, Mutation, OkResp, Op, P_ACTION, P_OVERWRITE, SaveResp, UploadResp,
};

/// Upper bound for the in-memory text endpoint (preview, later editor).
pub(super) const MAX_TEXT_BYTES: u64 = 2 * 1024 * 1024;

// ---------------------------------------------------------------------------
// Query params
// ---------------------------------------------------------------------------

/// Query params for `GET /api/files/{root_id}/{*path}`. Without `action` the
/// route lists the directory; `?action=download|preview|content` serve the
/// item itself.
///
/// The field names are the shared [`P_ACTION`] / [`P_FORMAT`] constants.
/// `#[serde(rename)]` only takes a literal, so that link cannot be written
/// here; `tests::query_fields_are_the_shared_constants` pins it instead.
#[derive(Deserialize, Default)]
pub struct FileQuery {
    #[serde(default)]
    action: Option<String>,
    #[serde(default)]
    format: Option<String>,
}

// ---------------------------------------------------------------------------
// Listing
// ---------------------------------------------------------------------------

/// GET /api/files/{root_id}/{*path} — list a directory, or serve the item
/// itself via `?action=download|preview|content`.
pub async fn file_get(
    State(state): State<Arc<AppState>>,
    auth: AuthUser,
    path: AxumPath<(i64, String)>,
    query: AxumQuery<FileQuery>,
    headers: axum::http::HeaderMap,
) -> Result<Response, ApiError> {
    let (root_id, req_rel) = path.0;
    match query.action.as_deref() {
        Some(a) if a == api_types::ACTION_DOWNLOAD => {
            let range = range_header(&headers);
            download(
                state,
                auth,
                root_id,
                req_rel,
                query.format.as_deref(),
                range,
                ims_header(&headers),
            )
            .await
        }
        Some(a) if a == api_types::ACTION_PREVIEW => {
            preview(
                state,
                auth,
                root_id,
                req_rel,
                range_header(&headers),
                ims_header(&headers),
            )
            .await
        }
        Some(a) if a == api_types::ACTION_CONTENT => content(state, auth, root_id, req_rel).await,
        Some(a) if a == api_types::ACTION_THUMB => thumb(state, auth, root_id, req_rel).await,
        _ => {
            let json = list_inner(state, auth, root_id, req_rel).await?;
            Ok(json.into_response())
        }
    }
}

/// GET /api/files/{root_id} — list the root directory itself, or serve the
/// root item via `?action=download|preview|content` (the root of a *file*
/// share is the file itself).
///
/// Same thing as [`file_get`] with an empty path, so it delegates there.
pub async fn list_root(
    state: State<Arc<AppState>>,
    auth: AuthUser,
    path: AxumPath<i64>,
    query: AxumQuery<FileQuery>,
    headers: axum::http::HeaderMap,
) -> Result<Response, ApiError> {
    file_get(
        state,
        auth,
        AxumPath((path.0, String::new())),
        query,
        headers,
    )
    .await
}

fn range_header(headers: &axum::http::HeaderMap) -> Option<String> {
    headers
        .get(header::RANGE)
        .and_then(|v| v.to_str().ok())
        .map(str::to_string)
}

fn ims_header(headers: &axum::http::HeaderMap) -> Option<String> {
    headers
        .get(header::IF_MODIFIED_SINCE)
        .and_then(|v| v.to_str().ok())
        .map(str::to_string)
}

/// Caching policy for a served file.
///
/// `private` because the response depends on who asked. `no-cache`, not
/// `no-store`, so a client can revalidate a large media file and get a `304`
/// instead of re-downloading it.
const FILE_CACHE: &str = "private, no-cache";

/// An mtime (unix seconds) as an HTTP-date, the `Last-Modified` format.
/// None for an unknown mtime (0) and for a file written in the last two
/// seconds: whole-second dates cannot tell two writes in one second apart.
fn http_date(mtime: i64) -> Option<String> {
    if mtime <= 0 || mtime + 2 > chrono::Utc::now().timestamp() {
        return None;
    }
    chrono::DateTime::from_timestamp(mtime, 0)
        .map(|d| d.format("%a, %d %b %Y %H:%M:%S GMT").to_string())
}

/// True when the client's `If-Modified-Since` is at or after `mtime`.
/// HTTP-dates carry whole seconds, so the comparison is second-precision.
fn unmodified_since(ims: Option<&str>, mtime: i64) -> bool {
    chrono::DateTime::parse_from_rfc2822(ims.unwrap_or_default())
        .is_ok_and(|t| t.timestamp() >= mtime)
}

async fn list_inner(
    state: Arc<AppState>,
    auth: AuthUser,
    root_id: i64,
    req_rel: String,
) -> Result<Json<FilesResp>, ApiError> {
    // A file share's root is the file itself: there is nothing to list.
    if auth.share.as_ref().is_some_and(|s| s.is_file) {
        return Err(FsError::NotADirectory.into());
    }
    let root = find_root(&auth.roots, root_id)?;
    let server_root = state.root.clone();
    let root_rel = root.path.clone();
    // Resolve and list in one blocking hop: both are filesystem work.
    let (entries, truncated) = blocking(move || {
        let full = fs::resolve_path(&server_root, &root_rel, &req_rel)?;
        fs::list_dir(&full)
    })
    .await?;

    Ok(Json(FilesResp { entries, truncated }))
}

// ---------------------------------------------------------------------------
// download / preview / content (milestone 4)
// ---------------------------------------------------------------------------

/// `Content-Disposition` parameters for `name`: an ASCII `filename=` fallback
/// (non-ASCII and control bytes become `_`) plus the RFC 8187 `filename*=`
/// that every current browser reads. Never fails header validation.
fn disposition(kind: &str, name: &str) -> String {
    let ascii: String = name
        .chars()
        .map(|c| match c {
            '"' | '\\' => '_',
            c if c.is_ascii_graphic() || c == ' ' => c,
            _ => '_',
        })
        .collect();
    let mut enc = String::with_capacity(name.len() * 3);
    for b in name.bytes() {
        // attr-char per RFC 8187.
        if b.is_ascii_alphanumeric() || b"!#$&+-.^_`|~".contains(&b) {
            enc.push(b as char);
        } else {
            use std::fmt::Write as _;
            let _ = write!(enc, "%{b:02X}");
        }
    }
    format!("{kind}; filename=\"{ascii}\"; filename*=UTF-8''{enc}")
}

/// Resolve the requested item to an absolute path + metadata (blocking).
async fn resolve_item(
    state: &AppState,
    root: &RootRow,
    req_rel: &str,
    share: Option<&ShareRow>,
) -> Result<(std::path::PathBuf, String, bool, u64, i64), ApiError> {
    let (server_root, root_rel, rel) = (state.root.clone(), root.path.clone(), req_rel.to_string());
    // A file share's synthetic root *is* the file, so resolve it directly.
    let share_target = share.filter(|s| s.is_file).map(|s| s.target.clone());
    blocking(move || {
        let full = match share_target {
            Some(target) => fs::resolve_file(&server_root, &target)?,
            None => fs::resolve_path(&server_root, &root_rel, &rel)?,
        };
        let name = full
            .file_name()
            .map(|n| n.to_string_lossy().into_owned())
            .ok_or_else(|| FsError::Invalid("invalid path".to_string()))?;
        let meta = std::fs::metadata(&full).map_err(|_| FsError::NotFound)?;
        let mtime = fs::mtime_secs(&meta).unwrap_or(0);
        Ok::<_, FsError>((full, name, meta.is_dir(), meta.len(), mtime))
    })
    .await
}

/// `GET ...?action=download` — a single file as-is, a folder as an archive
/// (format chosen by the client).
async fn download(
    state: Arc<AppState>,
    auth: AuthUser,
    root_id: i64,
    req_rel: String,
    format: Option<&str>,
    range: Option<String>,
    ims: Option<String>,
) -> Result<Response, ApiError> {
    let root = find_root(&auth.roots, root_id)?;
    let (full, name, is_dir, size, mtime) =
        resolve_item(&state, root, &req_rel, auth.share.as_ref()).await?;

    if !is_dir {
        return file_response(
            &full,
            &name,
            size,
            false,
            range.as_deref(),
            mtime,
            ims.as_deref(),
        )
        .await;
    }

    let fmt = format.and_then(ArchiveFormat::parse).ok_or_else(|| {
        ApiError::localized(
            StatusCode::BAD_REQUEST,
            "format must be one of: zip, tar, tar.gz, tar.zst",
            "err_bad_format",
        )
    })?;
    let disp = disposition("attachment", &format!("{name}.{}", fmt.extension()));
    let body = stream_archive(fmt, full, name);
    Response::builder()
        .status(StatusCode::OK)
        .header(header::CONTENT_TYPE, fmt.mime())
        .header(header::CONTENT_DISPOSITION, disp)
        .header(header::CACHE_CONTROL, FILE_CACHE)
        .body(body)
        .map_err(|e| {
            ApiError::new(
                StatusCode::INTERNAL_SERVER_ERROR,
                format!("bad response: {e}"),
            )
        })
}

/// `GET ...?action=preview` — a single file, inline (for native media).
async fn preview(
    state: Arc<AppState>,
    auth: AuthUser,
    root_id: i64,
    req_rel: String,
    range: Option<String>,
    ims: Option<String>,
) -> Result<Response, ApiError> {
    let root = find_root(&auth.roots, root_id)?;
    let (full, name, is_dir, size, mtime) =
        resolve_item(&state, root, &req_rel, auth.share.as_ref()).await?;
    if is_dir {
        return Err(ApiError::localized(
            StatusCode::BAD_REQUEST,
            "not a file",
            "err_not_a_file",
        ));
    }
    file_response(
        &full,
        &name,
        size,
        true,
        range.as_deref(),
        mtime,
        ims.as_deref(),
    )
    .await
}

/// `GET ...?action=content` — raw file bytes for the text preview/editor.
/// Capped at `MAX_TEXT_BYTES`.
async fn content(
    state: Arc<AppState>,
    auth: AuthUser,
    root_id: i64,
    req_rel: String,
) -> Result<Response, ApiError> {
    let root = find_root(&auth.roots, root_id)?;
    let (full, _name, is_dir, size, _mtime) =
        resolve_item(&state, root, &req_rel, auth.share.as_ref()).await?;
    if is_dir {
        return Err(ApiError::localized(
            StatusCode::BAD_REQUEST,
            "not a file",
            "err_not_a_file",
        ));
    }
    if size > MAX_TEXT_BYTES {
        return Err(ApiError::localized(
            StatusCode::PAYLOAD_TOO_LARGE,
            "file too large to preview",
            "err_too_large_preview",
        ));
    }
    // mtime and bytes from one handle, mtime first: a write in between would
    // otherwise hand the editor a stale conflict anchor for fresh content.
    let (mtime, bytes) = blocking(move || -> io::Result<(i64, Vec<u8>)> {
        let mut f = std::fs::File::open(&full)?;
        let mtime = fs::mtime_secs(&f.metadata()?).unwrap_or(0);
        let mut bytes = Vec::with_capacity(size as usize);
        f.read_to_end(&mut bytes)?;
        Ok((mtime, bytes))
    })
    .await?;
    Ok((
        [
            (
                header::CONTENT_TYPE,
                "text/plain; charset=utf-8".to_string(),
            ),
            (header::CACHE_CONTROL, FILE_CACHE.to_string()),
            (
                axum::http::HeaderName::from_static("x-file-mtime"),
                mtime.to_string(),
            ),
        ],
        bytes,
    )
        .into_response())
}

/// `GET ...?action=thumb` — a small WebP preview of an image or video.
///
/// `404` covers every "no thumbnail here" case: thumbnails switched off, a
/// folder, a kind we do not render, an undecodable file, a video without
/// ffmpeg. The grid falls back to its icon for all of them alike.
async fn thumb(
    state: Arc<AppState>,
    auth: AuthUser,
    root_id: i64,
    req_rel: String,
) -> Result<Response, ApiError> {
    let Some(thumbs) = state.thumbs.as_ref() else {
        return Err(no_thumb());
    };
    let root = find_root(&auth.roots, root_id)?;
    let (full, _name, is_dir, size, mtime) =
        resolve_item(&state, root, &req_rel, auth.share.as_ref()).await?;
    if is_dir {
        return Err(no_thumb());
    }
    // The kind is sniffed from the bytes, not the name, so an mp4 called .txt
    // still gets a thumbnail and a .jpg full of text does not.
    let kind = {
        let full = full.clone();
        blocking(move || Ok::<_, FsError>(fs::detect_kind(&full, false))).await?
    };
    let Some(bytes) = thumbs.get(&full, kind, size, mtime).await else {
        return Err(no_thumb());
    };
    Ok((
        [
            (header::CONTENT_TYPE, "image/webp"),
            // Safe despite the stable path: the client varies the query on
            // mtime, so new content always means a new URL.
            (
                header::CACHE_CONTROL,
                "private, max-age=31536000, immutable",
            ),
        ],
        bytes,
    )
        .into_response())
}

/// The message never reaches a user: an `<img>` 404 just leaves the tile's
/// icon showing. That is why it carries no localized code.
fn no_thumb() -> ApiError {
    ApiError::new(StatusCode::NOT_FOUND, "no thumbnail".to_string())
}

/// `PUT ...?action=content` — save a file's text contents (the editor).
///
/// Requires a read-write root. The body is the new contents. If the
/// `X-Expected-Mtime` header is present, the file's current mtime must match
/// it, otherwise `409 Conflict` (the file changed on disk since it was read).
/// Returns the file's new mtime so the client can anchor the next check.
pub async fn file_put(
    State(state): State<Arc<AppState>>,
    auth: AuthUser,
    path: AxumPath<(i64, String)>,
    query: AxumQuery<FileQuery>,
    headers: axum::http::HeaderMap,
    body: axum::body::Bytes,
) -> Result<Json<SaveResp>, ApiError> {
    let (root_id, req_rel) = path.0;
    put_inner(state, auth, root_id, req_rel, query, headers, body).await
}

/// `PUT .../{root_id}?action=content` — the root item itself. Only reachable
/// for a *file* share (its root is the file); for folders it resolves to a
/// directory and is rejected below.
pub async fn file_put_root(
    State(state): State<Arc<AppState>>,
    auth: AuthUser,
    path: AxumPath<i64>,
    query: AxumQuery<FileQuery>,
    headers: axum::http::HeaderMap,
    body: axum::body::Bytes,
) -> Result<Json<SaveResp>, ApiError> {
    put_inner(state, auth, path.0, String::new(), query, headers, body).await
}

async fn put_inner(
    state: Arc<AppState>,
    auth: AuthUser,
    root_id: i64,
    req_rel: String,
    query: AxumQuery<FileQuery>,
    headers: axum::http::HeaderMap,
    body: axum::body::Bytes,
) -> Result<Json<SaveResp>, ApiError> {
    if query.action.as_deref() != Some(api_types::ACTION_CONTENT) {
        return Err(ApiError::localized(
            StatusCode::BAD_REQUEST,
            "expected action=content",
            "err_bad_action",
        ));
    }
    let root = require_rw_root(&auth.roots, root_id)?;
    if body.len() as u64 > MAX_TEXT_BYTES {
        return Err(ApiError::localized(
            StatusCode::PAYLOAD_TOO_LARGE,
            "file too large to save",
            "err_too_large_save",
        ));
    }
    let expected: Option<i64> = headers
        .get("x-expected-mtime")
        .and_then(|v| v.to_str().ok())
        .and_then(|s| s.parse().ok());
    // A file share's synthetic root *is* the file, so resolve it directly.
    let share_target = auth
        .share
        .as_ref()
        .filter(|s| s.is_file)
        .map(|s| s.target.clone());
    let (server_root, root_rel, rel) = (state.root.clone(), root.path.clone(), req_rel);
    let content = body.to_vec();
    let mtime = blocking(move || match share_target {
        Some(target) => fs::save_file_at(&server_root, &target, &content, expected),
        None => fs::save_file(&server_root, &root_rel, &rel, &content, expected),
    })
    .await?;
    Ok(Json(SaveResp { mtime }))
}

/// Stream a single file to the client with the right disposition.
async fn file_response(
    full: &std::path::Path,
    name: &str,
    size: u64,
    inline: bool,
    range: Option<&str>,
    mtime: i64,
    ims: Option<&str>,
) -> Result<Response, ApiError> {
    let last_modified = http_date(mtime);
    // A revalidating client gets the empty 304 instead of the whole file. The
    // policy is repeated on it, so the stored copy does not lose the directive.
    if last_modified.is_some() && unmodified_since(ims, mtime) {
        return Ok((
            StatusCode::NOT_MODIFIED,
            [(header::CACHE_CONTROL, FILE_CACHE)],
        )
            .into_response());
    }
    let mime = mime_guess::from_path(full)
        .first_or_octet_stream()
        .to_string();
    let disp = disposition(if inline { "inline" } else { "attachment" }, name);
    // A single `bytes=a-b` range (media seeking). Anything else is served whole.
    let (start, end) = match parse_range(range, size) {
        Some(Some(r)) => r,
        Some(None) => {
            return Response::builder()
                .status(StatusCode::RANGE_NOT_SATISFIABLE)
                .header(header::CONTENT_RANGE, format!("bytes */{size}"))
                .body(axum::body::Body::empty())
                .map_err(|e| {
                    ApiError::new(
                        StatusCode::INTERNAL_SERVER_ERROR,
                        format!("bad response: {e}"),
                    )
                });
        }
        None => (0, size),
    };
    let partial = (start, end) != (0, size);
    let body = stream_file(full.to_path_buf(), start, end);
    let mut res = Response::builder()
        .status(if partial {
            StatusCode::PARTIAL_CONTENT
        } else {
            StatusCode::OK
        })
        .header(header::CONTENT_DISPOSITION, disp)
        .header(header::ACCEPT_RANGES, "bytes")
        .header(header::CONTENT_LENGTH, end - start)
        // Always sent, validator or not: with no directive a cache may apply
        // heuristic freshness to a response that depends on who asked.
        .header(header::CACHE_CONTROL, FILE_CACHE);
    if let Some(lm) = last_modified {
        // The validator the `no-cache` above revalidates against.
        res = res.header(header::LAST_MODIFIED, lm);
    }
    if partial {
        res = res.header(
            header::CONTENT_RANGE,
            format!("bytes {start}-{}/{size}", end - 1),
        );
    }
    // A file the browser would parse as a document (HTML/SVG/XML) is served
    // under the sandboxed policy, so it can render as a page without being
    // able to act as the app. Derived from the same `mime` we declare.
    // Non-scriptable inline files (PDF, …) are frameable by the app itself,
    // for the preview modal.
    if crate::api::is_scriptable_mime(&mime) {
        res = res.header("content-security-policy", crate::api::FILE_CSP);
    } else if inline {
        res = res
            .header("content-security-policy", crate::api::INLINE_CSP)
            .header(header::X_FRAME_OPTIONS, "SAMEORIGIN");
    }
    res.header(header::CONTENT_TYPE, mime)
        .body(body)
        .map_err(|e| {
            ApiError::new(
                StatusCode::INTERNAL_SERVER_ERROR,
                format!("bad response: {e}"),
            )
        })
}

/// Parse a `Range` header against `size`. `None` = serve the whole file,
/// `Some(None)` = unsatisfiable, `Some(Some((start, end)))` = half-open range.
fn parse_range(range: Option<&str>, size: u64) -> Option<Option<(u64, u64)>> {
    let spec = range?.strip_prefix("bytes=")?;
    // ponytail: one range only; multipart/byteranges is not worth it here.
    let (a, b) = spec.split_once('-')?;
    let (start, end) = match (a.trim().parse::<u64>().ok(), b.trim().parse::<u64>().ok()) {
        (Some(s), Some(e)) => (s, e.saturating_add(1).min(size)),
        (Some(s), None) if b.trim().is_empty() => (s, size),
        // Suffix form: the last N bytes.
        (None, Some(n)) if a.trim().is_empty() => (size.saturating_sub(n), size),
        _ => return None,
    };
    if start >= size || start >= end {
        return Some(None);
    }
    Some(Some((start, end)))
}

/// Stream `path[start..end)` to the client in chunks (blocking reader → channel).
fn stream_file(path: std::path::PathBuf, start: u64, end: u64) -> axum::body::Body {
    use std::io::Seek;
    let (tx, rx) = mpsc::channel::<Vec<u8>>(16);
    tokio::task::spawn_blocking(move || {
        let mut f = match std::fs::File::open(&path) {
            Ok(f) => f,
            Err(e) => {
                tracing::warn!(error = %e, path = %path.display(), "download open failed");
                return;
            }
        };
        if start > 0 && f.seek(io::SeekFrom::Start(start)).is_err() {
            return;
        }
        let mut left = end - start;
        let mut buf = vec![0u8; 256 * 1024];
        while left > 0 {
            let want = buf.len().min(left as usize);
            match f.read(&mut buf[..want]) {
                Ok(0) => break,
                Ok(n) => {
                    left -= n as u64;
                    // Client gone → stop producing.
                    if tx.blocking_send(buf[..n].to_vec()).is_err() {
                        break;
                    }
                }
                Err(e) => {
                    tracing::warn!(error = %e, path = %path.display(), "download read failed");
                    break;
                }
            }
        }
    });
    let stream = ReceiverStream::new(rx).map(Ok::<_, io::Error>);
    axum::body::Body::from_stream(stream)
}

/// Stream an archive of `dir` (top-level entry `top`) to the client.
fn stream_archive(fmt: ArchiveFormat, dir: std::path::PathBuf, top: String) -> axum::body::Body {
    let (tx, rx) = mpsc::channel::<Vec<u8>>(16);
    tokio::task::spawn_blocking(move || {
        let mut sink = ChanWriter::new(tx);
        if let Err(e) = archive::build(fmt, &dir, &top, &mut sink) {
            tracing::warn!(error = %e, dir = %dir.display(), "archive build failed");
        }
        // Dropping the sink flushes its buffer and closes the channel.
    });
    let stream = ReceiverStream::new(rx).map(Ok::<_, io::Error>);
    axum::body::Body::from_stream(stream)
}

/// A `Write` that buffers chunks and forwards them over an mpsc channel — the
/// bridge between the blocking archive builder and the async response body.
struct ChanWriter {
    tx: mpsc::Sender<Vec<u8>>,
    buf: Vec<u8>,
}

impl ChanWriter {
    fn new(tx: mpsc::Sender<Vec<u8>>) -> Self {
        Self {
            tx,
            buf: Vec::with_capacity(64 * 1024),
        }
    }
}

impl io::Write for ChanWriter {
    fn write(&mut self, b: &[u8]) -> io::Result<usize> {
        self.buf.extend_from_slice(b);
        if self.buf.len() >= 64 * 1024 {
            io::Write::flush(self)?;
        }
        Ok(b.len())
    }
    fn flush(&mut self) -> io::Result<()> {
        if !self.buf.is_empty() {
            let chunk = std::mem::take(&mut self.buf);
            self.tx
                .blocking_send(chunk)
                .map_err(|_| io::Error::new(io::ErrorKind::BrokenPipe, "client disconnected"))?;
        }
        Ok(())
    }
}

impl Drop for ChanWriter {
    fn drop(&mut self) {
        let _ = io::Write::flush(self);
    }
}

// ---------------------------------------------------------------------------
// POST dispatch: mkdir | rename/move/copy | upload
// ---------------------------------------------------------------------------

/// POST /api/files/{root_id} — top-level operations (upload into the root
/// directory). The bare root never targets a specific item, so JSON ops with
/// a missing folder name are rejected by the individual handlers.
pub async fn dispatch_root(
    State(state): State<Arc<AppState>>,
    auth: AuthUser,
    path: AxumPath<i64>,
    headers: axum::http::HeaderMap,
    req: axum::http::Request<axum::body::Body>,
) -> Result<Response, ApiError> {
    dispatch_inner(state, auth, path.0, String::new(), headers, req).await
}

/// POST /api/files/{root_id}/{*path}
pub async fn dispatch(
    State(state): State<Arc<AppState>>,
    auth: AuthUser,
    path: AxumPath<(i64, String)>,
    headers: axum::http::HeaderMap,
    req: axum::http::Request<axum::body::Body>,
) -> Result<Response, ApiError> {
    let (root_id, req_rel) = path.0;
    dispatch_inner(state, auth, root_id, req_rel, headers, req).await
}

/// Route one POST to upload, mutation or mkdir.
///
/// Upload and mutation are recognized by their content type. mkdir carries no
/// body, so it names itself with `?action=mkdir`. Anything else is rejected:
/// an unrecognized content type used to fall through to mkdir, which turned a
/// typo in a header into a silently created folder.
async fn dispatch_inner(
    state: Arc<AppState>,
    auth: AuthUser,
    root_id: i64,
    req_rel: String,
    headers: axum::http::HeaderMap,
    req: axum::http::Request<axum::body::Body>,
) -> Result<Response, ApiError> {
    let ct = headers
        .get(header::CONTENT_TYPE)
        .and_then(|v| v.to_str().ok())
        .unwrap_or("");

    if ct.starts_with("multipart/form-data") {
        return upload(state, auth, root_id, req_rel, req).await;
    }
    if ct.starts_with("application/json") {
        let bytes = axum::body::to_bytes(req.into_body(), 1_000_000)
            .await
            .map_err(|_| {
                ApiError::localized(
                    StatusCode::BAD_REQUEST,
                    "invalid request body",
                    "err_bad_body",
                )
            })?;
        let body: Mutation = axum::Json::from_bytes(&bytes)
            .map_err(|_| {
                ApiError::localized(
                    StatusCode::BAD_REQUEST,
                    "invalid request body",
                    "err_bad_body",
                )
            })?
            .0;
        return Ok(mutation(state, auth, root_id, req_rel, body)
            .await?
            .into_response());
    }
    match action_param(req.uri()).as_deref() {
        Some(api_types::ACTION_MKDIR) => {
            return Ok(mkdir(state, auth, root_id, req_rel).await?.into_response());
        }
        Some(api_types::ACTION_CREATE_FILE) => {
            return Ok(create_file(state, auth, root_id, req_rel)
                .await?
                .into_response());
        }
        _ => {}
    }
    Err(ApiError::localized(
        StatusCode::UNSUPPORTED_MEDIA_TYPE,
        "POST expects a multipart upload, a JSON mutation, or ?action=mkdir",
        "err_bad_post",
    ))
}

// ---------------------------------------------------------------------------
// create file
// ---------------------------------------------------------------------------

/// Create an empty file in a writable root.
async fn create_file(
    state: Arc<AppState>,
    auth: AuthUser,
    root_id: i64,
    req_rel: String,
) -> Result<Json<OkResp>, ApiError> {
    let root = require_rw_root(&auth.roots, root_id)?;
    if req_rel.trim().is_empty() {
        return Err(ApiError::localized(
            StatusCode::BAD_REQUEST,
            "a file name is required",
            "err_file_name_required",
        ));
    }
    let (server_root, root_rel, rel) = (state.root.clone(), root.path.clone(), req_rel);
    blocking(move || fs::create_file(&server_root, &root_rel, &rel)).await?;
    Ok(Json(OkResp {}))
}

// ---------------------------------------------------------------------------
// mkdir
// ---------------------------------------------------------------------------

async fn mkdir(
    state: Arc<AppState>,
    auth: AuthUser,
    root_id: i64,
    req_rel: String,
) -> Result<Json<OkResp>, ApiError> {
    let root = require_rw_root(&auth.roots, root_id)?;
    if req_rel.trim().is_empty() {
        return Err(ApiError::localized(
            StatusCode::BAD_REQUEST,
            "a folder name is required",
            "err_folder_name_required",
        ));
    }
    let (server_root, root_rel, rel) = (state.root.clone(), root.path.clone(), req_rel);
    blocking(move || fs::mkdir(&server_root, &root_rel, &rel)).await?;
    Ok(Json(OkResp {}))
}

// ---------------------------------------------------------------------------
// rename / move / copy
// ---------------------------------------------------------------------------

async fn mutation(
    state: Arc<AppState>,
    auth: AuthUser,
    root_id: i64,
    req_rel: String,
    body: Mutation,
) -> Result<Json<OkResp>, ApiError> {
    match body.op {
        Op::Rename => {
            let new_name = body
                .new_name
                .as_deref()
                .ok_or_else(|| {
                    ApiError::localized(
                        StatusCode::BAD_REQUEST,
                        "new_name is required",
                        "err_new_name_required",
                    )
                })?
                .to_string();
            let root = require_rw_root(&auth.roots, root_id)?;
            let (server_root, root_rel, rel, overwrite) = (
                state.root.clone(),
                root.path.clone(),
                req_rel,
                body.overwrite,
            );
            let vacated = blocking(move || {
                fs::rename_item(&server_root, &root_rel, &rel, &new_name, overwrite)
            })
            .await?;
            revoke_shares_at(&state, &vacated).await;
            Ok(Json(OkResp {}))
        }
        Op::Move | Op::Copy => {
            let dst_root_id = body.dst_root_id.ok_or_else(|| {
                ApiError::localized(
                    StatusCode::BAD_REQUEST,
                    "dst_root_id is required",
                    "err_dst_required",
                )
            })?;
            let dst = body.dst.clone().unwrap_or_default();
            // Moving or copying out of a folder requires rw there; copying
            // *from* a read-only root is fine.
            let op_is_move = body.op == Op::Move;
            let src_root = if op_is_move {
                require_rw_root(&auth.roots, root_id)?
            } else {
                find_root(&auth.roots, root_id)?
            };
            let dst_root = require_rw_root(&auth.roots, dst_root_id)?;
            let (server_root, src_rel, dst_rel, dst_path, rel, overwrite) = (
                state.root.clone(),
                src_root.path.clone(),
                dst_root.path.clone(),
                dst,
                req_rel,
                body.overwrite,
            );
            let vacated = blocking(move || {
                if op_is_move {
                    fs::move_item(&server_root, &src_rel, &rel, &dst_rel, &dst_path, overwrite)
                        .map(Some)
                } else {
                    // A copy frees no path, so it revokes nothing.
                    fs::copy_item(&server_root, &src_rel, &rel, &dst_rel, &dst_path, overwrite)
                        .map(|()| None)
                }
            })
            .await?;
            if let Some(vacated) = vacated {
                revoke_shares_at(&state, &vacated).await;
            }
            Ok(Json(OkResp {}))
        }
    }
}

// ---------------------------------------------------------------------------
// DELETE
// ---------------------------------------------------------------------------

pub async fn delete(
    State(state): State<Arc<AppState>>,
    auth: AuthUser,
    path: AxumPath<(i64, String)>,
) -> Result<Json<DeleteResp>, ApiError> {
    let (root_id, req_rel) = path.0;
    let root = require_rw_root(&auth.roots, root_id)?;
    if req_rel.trim().is_empty() {
        return Err(ApiError::localized(
            StatusCode::BAD_REQUEST,
            "a path inside the folder is required",
            "err_path_required",
        ));
    }
    let (server_root, root_rel, rel) = (state.root.clone(), root.path.clone(), req_rel);
    let (is_dir, gone) = blocking(move || fs::remove_item(&server_root, &root_rel, &rel)).await?;
    revoke_shares_at(&state, &gone).await;
    Ok(Json(DeleteResp { is_dir }))
}

// ---------------------------------------------------------------------------
// Upload (multipart)
// ---------------------------------------------------------------------------

async fn upload(
    state: Arc<AppState>,
    auth: AuthUser,
    root_id: i64,
    req_rel: String,
    req: axum::http::Request<axum::body::Body>,
) -> Result<Response, ApiError> {
    let root = require_rw_root(&auth.roots, root_id)?;
    // `base` is the upload directory; `root_abs` the user's root, which is the
    // containment boundary (a symlink may legitimately point elsewhere inside it).
    let (root_abs, base) = {
        let (server_root, root_rel, rel) = (state.root.clone(), root.path.clone(), req_rel);
        blocking(move || {
            Ok::<_, FsError>((
                fs::resolve_root(&server_root, &root_rel)?,
                fs::resolve_dir(&server_root, &root_rel, &rel)?,
            ))
        })
        .await?
    };
    let boundary = req
        .headers()
        .get(header::CONTENT_TYPE)
        .and_then(|v| v.to_str().ok())
        .and_then(parse_boundary)
        .ok_or_else(|| {
            ApiError::localized(
                StatusCode::BAD_REQUEST,
                "expected multipart/form-data with a boundary",
                "err_bad_multipart",
            )
        })?;
    let overwrite = parse_overwrite(req.uri());

    let stream = req.into_body().into_data_stream();
    let mut multipart = Multipart::new(stream, boundary);
    let mut uploaded: usize = 0;
    let mut skipped: Vec<String> = Vec::new();

    while let Some(mut field) = multipart.next_field().await.map_err(|_| {
        ApiError::localized(
            StatusCode::BAD_REQUEST,
            "invalid upload data",
            "err_bad_upload",
        )
    })? {
        let part_name = field
            .name()
            .filter(|n| !n.is_empty())
            .or_else(|| field.file_name())
            .map(str::to_string)
            .ok_or_else(|| {
                ApiError::localized(
                    StatusCode::BAD_REQUEST,
                    "part without a name",
                    "err_part_no_name",
                )
            })?;

        validate_rel_path(&part_name)?;

        let target = base.join(&part_name);
        let parent = target
            .parent()
            .filter(|p| !p.as_os_str().is_empty())
            .ok_or_else(|| {
                ApiError::localized(
                    StatusCode::BAD_REQUEST,
                    "invalid part name",
                    "err_bad_part_name",
                )
            })?;
        // Create the parent, then canonicalize it and require it to still be
        // inside the upload directory. `validate_rel_path` blocks `..`, but a
        // symlinked directory on disk would otherwise carry the write outside.
        let (p, b) = (parent.to_path_buf(), root_abs.clone());
        let file_name = target.file_name().map(|n| n.to_owned());
        let (parent, target_state) = tokio::task::spawn_blocking(move || {
            let escape = || io::Error::other("upload parent escapes the root");
            // Check the nearest existing ancestor *before* creating anything,
            // so no directory is ever created outside the root either.
            let mut existing = p.as_path();
            while !existing.exists() {
                existing = existing.parent().ok_or_else(escape)?;
            }
            if !fs::is_within_or_eq(&b, &existing.canonicalize()?) {
                return Err(escape());
            }
            if !p.is_dir() {
                std::fs::create_dir_all(&p)?;
            }
            let canon = p.canonicalize()?;
            if !fs::is_within_or_eq(&b, &canon) {
                return Err(escape());
            }
            // Whether the target already exists, and as what: two stats that
            // belong on this thread, not on an async worker.
            let state = file_name
                .map(|n| canon.join(n))
                .map(|t| (t.exists(), t.is_dir()));
            Ok::<_, io::Error>((canon, state))
        })
        .await
        .map_err(|_| io::Error::other("join"))?
        .map_err(|e| {
            tracing::warn!(error = %e, "upload parent rejected");
            ApiError::localized(
                StatusCode::FORBIDDEN,
                "invalid file path in upload",
                "err_bad_upload_path",
            )
        })?;
        let (Some(file_name), Some((exists, is_dir))) = (target.file_name(), target_state) else {
            return Err(ApiError::localized(
                StatusCode::BAD_REQUEST,
                "invalid part name",
                "err_bad_part_name",
            ));
        };
        let target = parent.join(file_name);

        if exists && !overwrite {
            // Drain this part and report it as a conflict at the end.
            while let Some(_chunk) = field.chunk().await.map_err(|_| {
                ApiError::localized(
                    StatusCode::BAD_REQUEST,
                    "invalid upload data",
                    "err_bad_upload",
                )
            })? {}
            skipped.push(part_name);
            continue;
        }
        if exists {
            if is_dir {
                return Err(ApiError::localized(
                    StatusCode::CONFLICT,
                    "a folder with this name already exists",
                    "err_folder_exists",
                ));
            }
            tokio::fs::remove_file(&target).await.map_err(|_| {
                ApiError::localized(
                    StatusCode::INTERNAL_SERVER_ERROR,
                    "internal error",
                    "err_internal",
                )
            })?;
        }

        // Stream to a temp file in the same directory, then rename into place.
        let suffix = crate::auth::random_token();
        let tmp = Scratch(parent.join(format!(".upload-{suffix}")));
        let tmp_file = tokio::fs::File::create(&tmp.0).await.map_err(|_| {
            ApiError::localized(
                StatusCode::INTERNAL_SERVER_ERROR,
                "internal error",
                "err_internal",
            )
        })?;
        // Buffered: a multipart chunk is often a few kilobytes, and each
        // unbuffered write would be its own syscall.
        let mut tmp_file = tokio::io::BufWriter::with_capacity(1 << 20, tmp_file);
        let write_failed = loop {
            match field.chunk().await.map_err(|_| {
                ApiError::localized(
                    StatusCode::BAD_REQUEST,
                    "invalid upload data",
                    "err_bad_upload",
                )
            }) {
                Ok(Some(chunk)) => {
                    if let Err(e) = tmp_file.write_all(&chunk).await {
                        tracing::warn!(error = %e, "write failed during upload");
                        break true;
                    }
                }
                Ok(None) => break tmp_file.flush().await.is_err(),
                Err(e) => return Err(e),
            }
        };
        if write_failed {
            return Err(ApiError::localized(
                StatusCode::INTERNAL_SERVER_ERROR,
                "could not save the file",
                "err_save_failed",
            ));
        }
        let tmp2 = tmp.0.clone();
        let target2 = target.clone();
        // Flatten both errors: the outer `Err` is a panicking or shut-down task,
        // the inner one is `rename` refusing. Either way nothing was published.
        let renamed = tokio::task::spawn_blocking(move || std::fs::rename(&tmp2, &target2))
            .await
            .map_err(io::Error::other)
            .and_then(|r| r);
        if let Err(e) = renamed {
            tracing::warn!(error = %e, path = %target.display(), "rename failed during upload");
            return Err(ApiError::localized(
                StatusCode::INTERNAL_SERVER_ERROR,
                "internal error",
                "err_internal",
            ));
        }
        tmp.disarm();
        uploaded += 1;
    }

    if uploaded == 0 && skipped.is_empty() {
        return Err(ApiError::localized(
            StatusCode::BAD_REQUEST,
            "no files were uploaded",
            "err_no_files_uploaded",
        ));
    }
    if !skipped.is_empty() {
        return Err(ApiError::localized(
            StatusCode::CONFLICT,
            "some files already exist",
            "err_files_exist",
        )
        .with_extra(serde_json::json!({ "skipped": skipped, "uploaded": uploaded })));
    }
    Ok(Json(UploadResp { uploaded }).into_response())
}

/// The `.upload-<token>` scratch file of one in-flight upload part. Dropping it
/// removes the file, which covers the paths no `return` can see, above all the
/// request future being dropped when the client closes the connection. A leaked
/// scratch file is never named again and shows up in listings, which include
/// hidden entries on purpose. [`Scratch::disarm`] after `rename` keeps the file.
struct Scratch(PathBuf);

impl Scratch {
    /// The scratch file is now the uploaded file: leave it alone.
    fn disarm(self) {
        std::mem::forget(self);
    }
}

impl Drop for Scratch {
    fn drop(&mut self) {
        // Plain blocking unlink: `Drop` can run during runtime shutdown, where
        // `tokio::spawn` panics. One unlink cannot block meaningfully.
        if let Err(e) = std::fs::remove_file(&self.0)
            && e.kind() != io::ErrorKind::NotFound
        {
            tracing::warn!(error = %e, path = %self.0.display(), "could not remove upload scratch file");
        }
    }
}

fn parse_boundary(content_type: &str) -> Option<String> {
    content_type
        .split(';')
        .map(|s| s.trim())
        .find_map(|s| s.strip_prefix("boundary="))
        .map(|b| b.trim_matches('"').to_string())
        .filter(|b| !b.is_empty())
}

/// Read one query parameter from the request URI.
fn query_param(uri: &axum::http::Uri, key: &str) -> Option<String> {
    uri.query()?.split('&').find_map(|kv| {
        let (k, v) = kv.split_once('=')?;
        (k == key).then(|| v.to_string())
    })
}

fn action_param(uri: &axum::http::Uri) -> Option<String> {
    query_param(uri, P_ACTION)
}

fn parse_overwrite(uri: &axum::http::Uri) -> bool {
    matches!(query_param(uri, P_OVERWRITE).as_deref(), Some("true" | "1"))
}

fn validate_rel_path(name: &str) -> Result<(), ApiError> {
    for c in std::path::Path::new(name).components() {
        match c {
            Component::Normal(_) => {}
            _ => {
                return Err(ApiError::localized(
                    StatusCode::BAD_REQUEST,
                    "invalid file path in upload",
                    "err_bad_upload_path",
                ));
            }
        }
    }
    Ok(())
}

// ---------------------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------------------

/// Drop every share that named `abs` or anything under it.
///
/// Called after a delete, a rename, or a move: each one frees a path, and a
/// share stores a path, not a file identity. Without this, a *new* item that
/// later lands on the freed path would inherit the old link's audience.
///
/// Only covers changes made through this API. A file moved out from under the
/// server (over SSH, say) leaves its shares in place, still pointing at a
/// path. Closing that needs inode pinning, which breaks across a restore from
/// backup, so it is deliberately not done.
///
/// Best-effort: the file operation has already succeeded by the time this
/// runs, so a database error must not turn it into a 500. The client would
/// read that as "the delete failed" and retry, and the retry would 404. The
/// failure is logged at `error` instead, and leaves a share pointing at a
/// path that no longer holds what it did.
pub(crate) async fn revoke_shares_at(state: &AppState, abs: &std::path::Path) {
    let target = target_rel(state, abs);
    match state.db.revoke_shares_at(&target).await {
        Ok(0) => {}
        Ok(n) => tracing::info!(target = %target, revoked = n, "shares revoked: path is gone"),
        Err(e) => {
            tracing::error!(error = %e, target = %target, "could not revoke shares on a freed path")
        }
    }
}

fn find_root(roots: &[RootRow], root_id: i64) -> Result<&RootRow, ApiError> {
    roots.iter().find(|r| r.id == root_id).ok_or_else(|| {
        ApiError::localized(
            StatusCode::FORBIDDEN,
            "no such folder",
            "err_no_such_folder",
        )
    })
}

fn require_rw_root(roots: &[RootRow], root_id: i64) -> Result<&RootRow, ApiError> {
    let root = find_root(roots, root_id)?;
    if !root.mode.is_writable() {
        return Err(ApiError::localized(
            StatusCode::FORBIDDEN,
            "read-only folder",
            "err_read_only_folder",
        ));
    }
    Ok(root)
}

#[cfg(test)]
mod tests {
    use super::*;
    use api_types::{ACTION_DOWNLOAD, P_FORMAT};
    use axum::http::Uri;

    /// A rename of `P_ACTION` or `P_FORMAT` without the matching field
    /// rename would silently stop the server from reading the parameter the
    /// client sends. This builds the query string from the constants and
    /// runs the real extractor over it.
    #[test]
    fn query_fields_are_the_shared_constants() {
        let uri: Uri = format!("/f/1/a.txt?{P_ACTION}={ACTION_DOWNLOAD}&{P_FORMAT}=zip")
            .parse()
            .unwrap();
        let q: FileQuery = AxumQuery::try_from_uri(&uri).unwrap().0;
        assert_eq!(q.action.as_deref(), Some(ACTION_DOWNLOAD));
        assert_eq!(q.format.as_deref(), Some("zip"));

        // The same constants drive the hand-rolled readers on the POST path.
        assert_eq!(action_param(&uri).as_deref(), Some(ACTION_DOWNLOAD));
        let uri: Uri = format!("/f/1/a.txt?{P_OVERWRITE}=true").parse().unwrap();
        assert!(parse_overwrite(&uri));
    }
}