defter 0.8.5

//! The Zotero Web API v3 subset a Zotero client syncs against, the read/query
//! API a CLI uses, and the parts of Zotero's local API that commonmeta sync
//! needs (single-object writes, children, the schema). See `docs/api.md`.
//!
//! Adapted from zhost (<https://github.com/mulatta/zhost>, MIT).

use std::collections::HashMap;
use std::sync::Arc;

use axum::{
    Extension, Form, Json, Router,
    body::{Body, Bytes},
    extract::{OriginalUri, Path, RawQuery, State},
    http::{HeaderMap, HeaderValue, StatusCode, header},
    response::{IntoResponse, Response},
    routing::{MethodRouter, delete, get, post},
};
use serde_json::{Map, Value, json};

use crate::app::{AppState, PendingDownload, PendingUpload};
use crate::auth;
use crate::control::LibraryRef;
use crate::error::AppError;
use crate::query::{ItemQuery, Params};
use crate::registry::Library;
use crate::store::{self, Kind, Outcome, Single};

/// The most objects a multi-object write may carry.
const MAX_WRITE_OBJECTS: usize = 50;

/// How long a fetched Zotero schema is served before it is revalidated.
const SCHEMA_MAX_AGE: i64 = 24 * 3600;

type Result<T> = std::result::Result<T, AppError>;

// ── routes ───────────────────────────────────────────────────────────────────

/// The library a data API request is about, which `auth::require_key` opened.
type Lib = Extension<Arc<Library>>;

/// Who a request's changes are attributed to: in a group library its user, as
/// `auth::require_key` found them; nobody in one's own.
#[derive(Debug, Clone, Default)]
pub struct Attribution(pub Option<store::Actor>);

type Who = Extension<Attribution>;

/// The data API under `prefix`, which requires a key.
pub fn data_routes(router: Router<AppState>, prefix: &str) -> Router<AppState> {
    let mut router = router.route(
        &format!("{prefix}/keys/current"),
        get(auth::key_current).delete(auth::key_delete),
    );
    router = router.route(&format!("{prefix}/groups/{{id}}"), get(group_get));
    for (path, _, methods) in library_routes() {
        router = router.route(&format!("{prefix}/users/{{id}}{path}"), methods.clone());
        router = router.route(&format!("{prefix}/groups/{{id}}{path}"), methods);
    }
    router
}

/// A route of a library: its path under `users/{id}`, the methods it answers
/// and their handlers.
pub type LibraryRoute = (
    &'static str,
    &'static [&'static str],
    MethodRouter<AppState>,
);

/// Every route of a library, a user's under `users/{id}` and a group's under
/// `groups/{id}`. The access tests go through this list, every method of every
/// route with every kind of caller, so a route cannot be left out of them, nor a
/// method answer that it does not list.
pub fn library_routes() -> Vec<LibraryRoute> {
    let objects = |kind: Kind| {
        get(
            move |State(app): State<AppState>,
                  Extension(lib): Lib,
                  OriginalUri(uri),
                  RawQuery(raw): RawQuery| {
                objects_get(app, lib, kind, uri.path().to_string(), raw)
            },
        )
        .post(
            move |Extension(lib): Lib, Extension(who): Who, headers: HeaderMap, body: Bytes| {
                objects_write(lib, who, kind, headers, body)
            },
        )
        .patch(
            move |Extension(lib): Lib, Extension(who): Who, headers: HeaderMap, body: Bytes| {
                objects_write(lib, who, kind, headers, body)
            },
        )
        .delete(
            move |State(app): State<AppState>,
                  Extension(lib): Lib,
                  headers: HeaderMap,
                  RawQuery(raw): RawQuery| {
                objects_delete(app, lib, kind, headers, raw)
            },
        )
    };
    let object = |kind: Kind| {
        get(
            move |Extension(lib): Lib, Path((_, key)): Path<(String, String)>| {
                object_get(lib, kind, key)
            },
        )
        .put(
            move |State(app): State<AppState>,
                  Extension(lib): Lib,
                  Extension(who): Who,
                  Path((_, key)): Path<(String, String)>,
                  headers: HeaderMap,
                  body: Bytes| {
                object_write(app, lib, who, kind, key, headers, body, false)
            },
        )
        .patch(
            move |State(app): State<AppState>,
                  Extension(lib): Lib,
                  Extension(who): Who,
                  Path((_, key)): Path<(String, String)>,
                  headers: HeaderMap,
                  body: Bytes| {
                object_write(app, lib, who, kind, key, headers, body, true)
            },
        )
        .delete(
            move |State(app): State<AppState>,
                  Extension(lib): Lib,
                  Path((_, key)): Path<(String, String)>,
                  headers: HeaderMap| { object_delete(app, lib, kind, key, headers) },
        )
    };

    let items = get(items_get)
        .post(
            |Extension(lib): Lib, Extension(who): Who, headers: HeaderMap, body: Bytes| {
                objects_write(lib, who, Kind::Item, headers, body)
            },
        )
        .patch(
            |Extension(lib): Lib, Extension(who): Who, headers: HeaderMap, body: Bytes| {
                objects_write(lib, who, Kind::Item, headers, body)
            },
        )
        .delete(
            |State(app): State<AppState>,
             Extension(lib): Lib,
             headers: HeaderMap,
             RawQuery(raw): RawQuery| {
                objects_delete(app, lib, Kind::Item, headers, raw)
            },
        );
    const READ: &[&str] = &["GET"];
    const OBJECTS: &[&str] = &["GET", "POST", "PATCH", "DELETE"];
    const OBJECT: &[&str] = &["GET", "PUT", "PATCH", "DELETE"];
    vec![
        ("/groups", &["GET", "POST"], get(groups).post(groups_mirror)),
        (
            "/settings",
            &["GET", "POST", "DELETE"],
            get(settings_get)
                .post(settings_post)
                .delete(settings_delete),
        ),
        ("/collections", OBJECTS, objects(Kind::Collection)),
        ("/collections/{key}", OBJECT, object(Kind::Collection)),
        ("/collections/{key}/items", READ, get(collection_items)),
        (
            "/collections/{key}/items/top",
            READ,
            get(collection_items_top),
        ),
        ("/searches", OBJECTS, objects(Kind::Search)),
        ("/searches/{key}", OBJECT, object(Kind::Search)),
        ("/items", OBJECTS, items),
        ("/items/top", READ, get(items_top)),
        ("/items/trash", READ, get(items_trash)),
        ("/items/{key}", OBJECT, object(Kind::Item)),
        ("/items/{key}/children", READ, get(item_children)),
        ("/items/{key}/fulltext", READ, get(fulltext_item)),
        (
            "/items/{key}/file",
            &["GET", "POST"],
            get(file_get).post(file_post),
        ),
        (
            "/tags",
            &["GET", "DELETE"],
            get(tags_get).delete(tags_delete),
        ),
        (
            "/fulltext",
            &["GET", "POST"],
            get(fulltext_versions).post(fulltext_write),
        ),
        ("/deleted", READ, get(deleted)),
    ]
}

/// Endpoints under `prefix` that need no key: the API root, the schema and
/// the login session.
pub fn public_routes(router: Router<AppState>, prefix: &str) -> Router<AppState> {
    let mut router = router
        .route(&format!("{prefix}/schema"), get(schema))
        .route(
            &format!("{prefix}/keys/sessions"),
            post(auth::create_session),
        )
        .route(
            &format!("{prefix}/keys/sessions/{{token}}"),
            get(auth::check_session).delete(auth::cancel_session),
        );
    if !prefix.is_empty() {
        router = router
            .route(prefix, get(api_root))
            .route(&format!("{prefix}/"), get(api_root));
    }
    router
}

/// Byte transfers, authorized by the unguessable token in their URL.
pub fn file_routes() -> Router<AppState> {
    Router::new()
        .route("/uploads/{token}", post(upload))
        .route("/downloads/{token}", get(download))
        .route("/uploads/{token}/cancel", delete(upload_cancel))
}

// ── helpers ──────────────────────────────────────────────────────────────────

fn version_headers(version: i64) -> HeaderMap {
    let mut headers = HeaderMap::new();
    headers.insert("last-modified-version", HeaderValue::from(version));
    headers
}

/// Header value from stored text, dropping a value HTTP cannot carry rather
/// than failing the response.
fn header_value(text: &str) -> HeaderValue {
    HeaderValue::from_str(text).unwrap_or_else(|_| HeaderValue::from_static(""))
}

/// The headers that identify this server, as Zotero's local API does.
pub fn server_headers(server_id: &str) -> HeaderMap {
    let mut headers = HeaderMap::new();
    headers.insert("zotero-server-id", header_value(server_id));
    headers.insert(
        "x-defter-version",
        HeaderValue::from_static(env!("CARGO_PKG_VERSION")),
    );
    headers
}

async fn current(lib: &Library) -> Result<i64> {
    lib.db.read(store::current_version).await
}

/// Record how many bytes `lib`'s attachment files take, for the admin page,
/// after files were added or removed. A failure costs only that overview.
pub(crate) async fn note_usage(app: &AppState, lib: &Library) {
    let counted = match lib.db.read(|c| store::file_usage(c, "")).await {
        Ok((bytes, _)) => app.control.note_usage(lib.id, bytes).await,
        Err(error) => Err(error),
    };
    if let Err(error) = counted {
        tracing::warn!(%error, library = %lib.id, "recording storage use");
    }
}

const MB: u64 = 1024 * 1024;

/// The refusal if `filesize` bytes for `item_key` would take the library's user
/// past their quota, as zotero.org answers it: 413 with the usage and the quota
/// in MB, which the client shows. A file it replaces frees its space.
pub(crate) async fn over_quota(
    app: &AppState,
    lib: &Library,
    item_key: &str,
    filesize: i64,
) -> Result<Option<Response>> {
    // A group's files count against its owner's quota, with their own.
    let payer = match lib.id {
        LibraryRef::User(id) => app.control.user(id).await?,
        LibraryRef::Group(id) => app.control.group_owner(id).await?,
    };
    let Some((id, quota)) = payer.and_then(|u| app.quota(&u).map(|q| (u.id, q))) else {
        return Ok(None);
    };
    let lookup = item_key.to_string();
    let (here, replaced) = lib.db.read(move |c| store::file_usage(c, &lookup)).await?;
    let used = here + app.control.usage_of(id, lib.id).await?;
    if (used - replaced).max(0) as u64 + filesize.max(0) as u64 <= quota {
        return Ok(None);
    }
    let mut headers = HeaderMap::new();
    headers.insert(
        "zotero-storage-usage",
        HeaderValue::from(used.max(0) as u64 / MB),
    );
    headers.insert("zotero-storage-quota", HeaderValue::from(quota / MB));
    headers.insert("zotero-storage-userid", HeaderValue::from(id));
    Ok(Some(
        (
            StatusCode::PAYLOAD_TOO_LARGE,
            headers,
            "File would exceed quota",
        )
            .into_response(),
    ))
}

fn header_i64(headers: &HeaderMap, name: &str) -> Option<i64> {
    headers
        .get(name)
        .and_then(|v| v.to_str().ok())
        .and_then(|s| s.trim().parse().ok())
}

fn if_unmodified(headers: &HeaderMap) -> Option<i64> {
    header_i64(headers, "if-unmodified-since-version")
}

/// The `If-Unmodified-Since-Version` a write must carry; without it the
/// version guard would be bypassed.
fn require_unmodified(headers: &HeaderMap) -> Result<i64> {
    if_unmodified(headers).ok_or_else(|| {
        AppError::PreconditionRequired("If-Unmodified-Since-Version required".into())
    })
}

fn conflict(version: i64) -> Response {
    (StatusCode::PRECONDITION_FAILED, version_headers(version)).into_response()
}

fn json_body<T: serde::de::DeserializeOwned>(body: &Bytes) -> Result<T> {
    serde_json::from_slice(body).map_err(|e| AppError::BadRequest(format!("malformed JSON: {e}")))
}

/// The Web API's write result: `successful` with the full objects (the client
/// rejects a partial echo), `success` with their keys.
fn write_result(result: store::BatchResult) -> Response {
    let success: Map<String, Value> = result
        .successful
        .iter()
        .map(|(i, o)| (i.clone(), o.get("key").cloned().unwrap_or(Value::Null)))
        .collect();
    (
        version_headers(result.version),
        Json(json!({
            "successful": result.successful,
            "success": success,
            "unchanged": {},
            "failed": result.failed,
        })),
    )
        .into_response()
}

/// `Link: <…>; rel="next"` for the page after `start`, keeping the request's
/// other parameters and pointing at the public URL.
fn next_link(public_url: &str, path: &str, raw: Option<&str>, start: i64) -> String {
    let mut pairs: Vec<(String, String)> = raw
        .and_then(|q| serde_urlencoded::from_str(q).ok())
        .unwrap_or_default();
    pairs.retain(|(k, _)| k != "start");
    pairs.push(("start".into(), start.to_string()));
    let qs = serde_urlencoded::to_string(&pairs).unwrap_or_default();
    format!("<{public_url}{path}?{qs}>; rel=\"next\"")
}

/// A page of objects with `Total-Results` and, while more remain, `Link`.
fn paged(
    app: &AppState,
    version: i64,
    path: &str,
    raw: Option<&str>,
    (items, total): (Vec<Value>, i64),
    (limit, start): (i64, i64),
) -> Response {
    let mut headers = version_headers(version);
    headers.insert("total-results", HeaderValue::from(total));
    if start + limit < total {
        headers.insert(
            header::LINK,
            header_value(&next_link(&app.cfg.public_url, path, raw, start + limit)),
        );
    }
    (headers, Json(Value::Array(items))).into_response()
}

// ── API root and schema ──────────────────────────────────────────────────────

/// `GET /api/`: identifies the server like Zotero's local API does, so
/// commonmeta can tell it is talking to a Zotero API.
async fn api_root(State(app): State<AppState>) -> Response {
    (
        server_headers(&app.server_id),
        Json(json!({ "name": "defter", "version": env!("CARGO_PKG_VERSION") })),
    )
        .into_response()
}

fn now() -> i64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs() as i64)
        .unwrap_or(0)
}

/// `GET /schema`: Zotero's item schema, which commonmeta sync reads to map
/// works onto item types.
async fn schema(State(app): State<AppState>) -> Result<Response> {
    let body = schema_json(&app).await?;
    Ok(([(header::CONTENT_TYPE, "application/json")], body).into_response())
}

/// Zotero's item schema as JSON: fetched from zotero.org, kept in the database
/// and revalidated daily; a stale copy is used when zotero.org is unreachable.
pub(crate) async fn schema_json(app: &AppState) -> Result<String> {
    let (cached, etag, fetched) = (
        app.control.meta("schema").await?,
        app.control.meta("schema_etag").await?,
        app.control
            .meta("schema_fetched")
            .await?
            .and_then(|s| s.parse::<i64>().ok())
            .unwrap_or(0),
    );
    if let Some(body) = &cached
        && now() - fetched < SCHEMA_MAX_AGE
    {
        return Ok(body.clone());
    }

    let mut request = app.http.get(&app.cfg.schema_url);
    if let (Some(etag), Some(_)) = (&etag, &cached) {
        request = request.header(header::IF_NONE_MATCH, etag);
    }
    let fetched = async {
        let resp = request.send().await?;
        if resp.status() == StatusCode::NOT_MODIFIED {
            return Ok(None);
        }
        let resp = resp.error_for_status()?;
        let etag = resp
            .headers()
            .get(header::ETAG)
            .and_then(|v| v.to_str().ok())
            .map(String::from);
        Ok::<_, reqwest::Error>(Some((resp.text().await?, etag)))
    }
    .await;

    match (fetched, cached) {
        (Ok(None), Some(body)) => {
            app.control
                .set_meta(vec![("schema_fetched", now().to_string())])
                .await?;
            Ok(body)
        }
        (Ok(Some((body, etag))), _) if serde_json::from_str::<Value>(&body).is_ok() => {
            app.control
                .set_meta(vec![
                    ("schema", body.clone()),
                    ("schema_etag", etag.unwrap_or_default()),
                    ("schema_fetched", now().to_string()),
                ])
                .await?;
            Ok(body)
        }
        (result, Some(body)) => {
            tracing::warn!(
                ?result,
                "Zotero schema refresh failed; serving the stored copy"
            );
            Ok(body)
        }
        (Err(e), None) => Err(AppError::Upstream(format!("Zotero schema: {e}"))),
        (Ok(_), None) => Err(AppError::Upstream(
            "Zotero schema: unusable response".into(),
        )),
    }
}

// ── library data ─────────────────────────────────────────────────────────────

/// `GET /users/<id>/groups`: the groups the user is a member of. With
/// `format=versions` a map of group ID to the group's version, which the client
/// compares with its own, otherwise the groups' JSON, which its Choose Libraries
/// dialog pages through. Never an empty answer for want of reading them: a
/// client takes a group missing here for one it was removed from, and offers to
/// delete its copy.
async fn groups(
    State(app): State<AppState>,
    Extension(lib): Lib,
    RawQuery(raw): RawQuery,
) -> Result<Response> {
    let LibraryRef::User(user_id) = lib.id else {
        return Err(AppError::NotFound("no such route".into()));
    };
    let versions = raw
        .as_deref()
        .unwrap_or("")
        .split('&')
        .any(|pair| pair == "format=versions");
    let groups = app.control.groups_of(user_id).await?;
    let body = if versions {
        Value::Object(
            groups
                .iter()
                .map(|(group, _)| (group.id.to_string(), json!(group.version)))
                .collect(),
        )
    } else {
        let mut list = Vec::new();
        for (group, _) in &groups {
            let Some((group, members)) = app.control.group_with_members(group.id).await? else {
                continue;
            };
            let library = app.libraries.get(LibraryRef::Group(group.id)).await?;
            list.push(group_json(&app, &group, &members, &library).await?);
        }
        Value::Array(list)
    };
    Ok((version_headers(current(&lib).await?), Json(body)).into_response())
}

/// `POST /users/<id>/groups`, with the user's zotero.org API key (the one
/// Zotero syncs with) in `Zotero-Org-API-Key`: mirror their groups from
/// zotero.org, where groups and their members are managed
/// ([`crate::control::Control::mirror_groups`]), then answer as `GET`. The key
/// must be the user's own; it is used for these requests and not kept. `400`
/// without one, `403` for a key zotero.org refuses or another user's, `502`
/// when zotero.org cannot be asked.
async fn groups_mirror(
    State(app): State<AppState>,
    Extension(lib): Lib,
    headers: HeaderMap,
    RawQuery(raw): RawQuery,
) -> Result<Response> {
    let LibraryRef::User(user_id) = lib.id else {
        return Err(AppError::NotFound("no such route".into()));
    };
    let key = headers
        .get("zotero-org-api-key")
        .and_then(|v| v.to_str().ok())
        .map(str::trim)
        .filter(|k| !k.is_empty())
        .ok_or_else(|| {
            AppError::BadRequest(
                "Zotero-Org-API-Key: the zotero.org API key Zotero syncs with".into(),
            )
        })?
        .to_string();
    use crate::zotero_org::Refusal;
    let forbidden = |why: &'static str| (StatusCode::FORBIDDEN, why).into_response();
    let api = app.cfg.zotero_api_url.clone();
    let account = match crate::zotero_org::account(&app.http, &api, &key).await {
        Ok(account) => account,
        Err(Refusal::Rejected) => return Ok(forbidden("zotero.org does not know this key")),
        Err(Refusal::Unreachable(detail)) => return Err(AppError::Upstream(detail)),
    };
    if account.user_id != user_id {
        return Ok(forbidden("the zotero.org key is another user's"));
    }
    let listed = match crate::zotero_org::groups(&app.http, &api, &key, user_id).await {
        Ok(listed) => listed,
        Err(Refusal::Rejected) => return Ok(forbidden("zotero.org does not know this key")),
        Err(Refusal::Unreachable(detail)) => return Err(AppError::Upstream(detail)),
    };
    let mirrored = listed
        .into_iter()
        .map(|g| crate::control::GroupMirror {
            id: g.id,
            name: g.name,
            description: g.description,
            library_editing: g.library_editing,
            file_editing: g.file_editing,
            owner: g.owner,
            admins: g.admins,
            members: g.members,
            kind: g.kind,
            library_reading: g.library_reading,
            url: g.url,
        })
        .collect();
    let changed = app.control.mirror_groups(user_id, mirrored).await?;
    tracing::info!(
        user = user_id,
        changed = changed.len(),
        "mirrored groups from zotero.org"
    );
    groups(State(app), Extension(lib), RawQuery(raw)).await
}

/// `GET /groups/<id>`: the group, as Zotero's API describes it, with the
/// group's own version.
async fn group_get(State(app): State<AppState>, Extension(lib): Lib) -> Result<Response> {
    let LibraryRef::Group(id) = lib.id else {
        return Err(AppError::NotFound("no such group".into()));
    };
    let (group, members) = app
        .control
        .group_with_members(id)
        .await?
        .ok_or_else(|| AppError::NotFound("no such group".into()))?;
    let body = group_json(&app, &group, &members, &lib).await?;
    Ok((version_headers(group.version), Json(body)).into_response())
}

/// A group's JSON, as zotero.org's: its settings, owner, admins and members by
/// user ID (the owner and admins are not repeated among the members), of
/// those on defter, as mirrored from zotero.org.
async fn group_json(
    app: &AppState,
    group: &crate::control::Group,
    members: &[crate::control::Member],
    library: &Library,
) -> Result<Value> {
    use crate::control::Role;
    let ids = |role: Role| -> Vec<u64> {
        members
            .iter()
            .filter(|m| m.role == role)
            .map(|m| m.user.id)
            .collect()
    };
    // zotero.org's owner, whether or not they are on defter.
    let owner = group.owner;
    let items = library.db.read(store::stats).await?.items;
    let created = crate::backup::iso_time(group.created_at.max(0) as u64);
    let url = format!("{}/groups/{}", app.cfg.public_url, group.id);
    Ok(json!({
        "id": group.id,
        "version": group.version,
        "links": {
            "self": { "href": url, "type": "application/json" },
            "alternate": { "href": app.cfg.public_url, "type": "text/html" },
        },
        "meta": {
            "created": created,
            "lastModified": created,
            "numItems": items,
        },
        "data": {
            "id": group.id,
            "version": group.version,
            "name": group.name,
            "owner": owner,
            "type": group.kind,
            "description": group.description,
            "url": group.url,
            "libraryEditing": group.library_editing,
            "libraryReading": group.library_reading,
            "fileEditing": group.file_editing,
            "admins": ids(Role::Admin),
            "members": ids(Role::Member),
        },
    }))
}

/// `GET /collections`, `GET /searches`: the `format=versions` map, a batch by
/// `<kind>Key`, or a page of all of them.
async fn objects_get(
    app: AppState,
    lib: Arc<Library>,
    kind: Kind,
    path: String,
    raw: Option<String>,
) -> Result<Response> {
    let params = Params::parse(raw.as_deref());
    let version = current(&lib).await?;
    if params.get("format") == Some("versions") {
        let since = params.since();
        // Always 200 with the (possibly empty) map, never 304: the client's
        // version read sends no If-Modified-Since-Version and loops on a 304.
        let map = lib
            .db
            .read(move |c| store::versions(c, kind, since, false))
            .await?;
        return Ok((version_headers(version), Json(map)).into_response());
    }
    let keys = params.csv(kind.key_param());
    if !keys.is_empty() {
        let objects = lib.db.read(move |c| store::objects(c, kind, &keys)).await?;
        return Ok((version_headers(version), Json(objects)).into_response());
    }
    let page = params.page();
    let listing = lib
        .db
        .read(move |c| store::list_objects(c, kind, page.0, page.1))
        .await?;
    Ok(paged(&app, version, &path, raw.as_deref(), listing, page))
}

/// `POST`/`PATCH` on a collection endpoint: create or update a batch.
async fn objects_write(
    lib: Arc<Library>,
    who: Attribution,
    kind: Kind,
    headers: HeaderMap,
    body: Bytes,
) -> Result<Response> {
    let batch: Vec<Value> = json_body(&body)?;
    if batch.len() > MAX_WRITE_OBJECTS {
        return Ok((
            StatusCode::PAYLOAD_TOO_LARGE,
            format!("at most {MAX_WRITE_OBJECTS} objects per request"),
        )
            .into_response());
    }
    let expected = if_unmodified(&headers);
    match lib
        .db
        .write(move |c| store::write_batch(c, kind, batch, expected, who.0.as_ref()))
        .await?
    {
        Outcome::Done(result) => Ok(write_result(result)),
        Outcome::Conflict(current) => Ok(conflict(current)),
    }
}

/// `DELETE /<kind>s?<kind>Key=a,b`.
async fn objects_delete(
    app: AppState,
    lib: Arc<Library>,
    kind: Kind,
    headers: HeaderMap,
    raw: Option<String>,
) -> Result<Response> {
    let expected = require_unmodified(&headers)?;
    let keys = Params::parse(raw.as_deref()).csv(kind.key_param());
    match lib
        .db
        .write(move |c| store::delete_batch(c, kind, &keys, Some(expected)))
        .await?
    {
        Outcome::Done((version, files)) => {
            let removed = !files.is_empty();
            for key in files {
                app.storage.remove(&lib.files, &key).await;
            }
            if removed {
                note_usage(&app, &lib).await;
            }
            Ok((StatusCode::NO_CONTENT, version_headers(version)).into_response())
        }
        Outcome::Conflict(current) => Ok(conflict(current)),
    }
}

/// `GET /<kind>s/<key>`, with the object's own version.
async fn object_get(lib: Arc<Library>, kind: Kind, key: String) -> Result<Response> {
    match lib.db.read(move |c| store::object(c, kind, &key)).await? {
        Some((version, object)) => Ok((version_headers(version), Json(object)).into_response()),
        None => Err(AppError::NotFound(format!("{} not found", kind.as_str()))),
    }
}

/// `PUT` (replace) or `PATCH` (merge) `/<kind>s/<key>`. The precondition is
/// the object's version, from `If-Unmodified-Since-Version` or the body.
#[allow(
    clippy::too_many_arguments,
    reason = "what a route's extractors hand over"
)]
async fn object_write(
    app: AppState,
    lib: Arc<Library>,
    who: Attribution,
    kind: Kind,
    key: String,
    headers: HeaderMap,
    body: Bytes,
    merge: bool,
) -> Result<Response> {
    let provided: Value = json_body(&body)?;
    let expected = if_unmodified(&headers)
        .or_else(|| provided.get("version").and_then(Value::as_i64))
        .ok_or_else(|| {
            AppError::PreconditionRequired(
                "If-Unmodified-Since-Version or a version property required".into(),
            )
        })?;
    let outcome = lib
        .db
        .write(move |c| store::write_one(c, kind, &key, provided, expected, merge, who.0.as_ref()))
        .await?;
    single_response(&app, &lib, outcome, None).await
}

/// `DELETE /<kind>s/<key>`.
async fn object_delete(
    app: AppState,
    lib: Arc<Library>,
    kind: Kind,
    key: String,
    headers: HeaderMap,
) -> Result<Response> {
    let expected = require_unmodified(&headers)?;
    let item_key = key.clone();
    let outcome = lib
        .db
        .write(move |c| store::delete_one(c, kind, &key, expected))
        .await?;
    single_response(&app, &lib, outcome, Some(&item_key)).await
}

async fn single_response(
    app: &AppState,
    lib: &Library,
    outcome: Single,
    item_key: Option<&str>,
) -> Result<Response> {
    match outcome {
        Single::Done {
            version,
            file_removed,
        } => {
            if file_removed && let Some(key) = item_key {
                app.storage.remove(&lib.files, key).await;
                note_usage(app, lib).await;
            }
            Ok((StatusCode::NO_CONTENT, version_headers(version)).into_response())
        }
        Single::Conflict(version) => Ok(conflict(version)),
        Single::NotFound => Err(AppError::NotFound("object not found".into())),
        Single::Invalid(message) => Err(AppError::BadRequest(message)),
    }
}

// ── items ────────────────────────────────────────────────────────────────────

/// The sync reads shared by `/items` and `/items/top`: the `format=versions`
/// map (top-level only with `top`) and the `itemKey` batch. `None` for a
/// listing query.
async fn item_sync_read(lib: &Library, params: &Params, top: bool) -> Result<Option<Response>> {
    if params.get("format") == Some("versions") {
        let since = params.since();
        let version = current(lib).await?;
        let map = lib
            .db
            .read(move |c| store::versions(c, Kind::Item, since, top))
            .await?;
        return Ok(Some((version_headers(version), Json(map)).into_response()));
    }
    let keys = params.csv("itemKey");
    if !keys.is_empty() {
        let version = current(lib).await?;
        let items = lib
            .db
            .read(move |c| store::objects(c, Kind::Item, &keys))
            .await?;
        return Ok(Some(
            (version_headers(version), Json(items)).into_response(),
        ));
    }
    Ok(None)
}

/// An item listing: every matching key as plain text with `format=keys` (what
/// the Zotero client reads when restoring a collection), a JSON page otherwise.
async fn item_listing(
    app: &AppState,
    lib: &Library,
    path: &str,
    raw: Option<&str>,
    params: &Params,
    q: ItemQuery,
) -> Result<Response> {
    let version = current(lib).await?;
    if params.get("format") == Some("keys") {
        let keys = lib.db.read(move |c| store::item_keys(c, &q)).await?;
        return Ok((version_headers(version), keys.join("\n")).into_response());
    }
    let page = (q.limit, q.start);
    let listing = lib.db.read(move |c| store::query_items(c, &q)).await?;
    Ok(paged(app, version, path, raw, listing, page))
}

async fn items_get(
    State(app): State<AppState>,
    Extension(lib): Lib,
    OriginalUri(uri): OriginalUri,
    RawQuery(raw): RawQuery,
) -> Result<Response> {
    let params = Params::parse(raw.as_deref());
    if let Some(response) = item_sync_read(&lib, &params, false).await? {
        return Ok(response);
    }
    let q = ItemQuery::from_params(&params);
    item_listing(&app, &lib, uri.path(), raw.as_deref(), &params, q).await
}

async fn items_top(
    State(app): State<AppState>,
    Extension(lib): Lib,
    OriginalUri(uri): OriginalUri,
    RawQuery(raw): RawQuery,
) -> Result<Response> {
    let params = Params::parse(raw.as_deref());
    if let Some(response) = item_sync_read(&lib, &params, true).await? {
        return Ok(response);
    }
    let mut q = ItemQuery::from_params(&params);
    q.top = true;
    item_listing(&app, &lib, uri.path(), raw.as_deref(), &params, q).await
}

async fn items_trash(
    State(app): State<AppState>,
    Extension(lib): Lib,
    OriginalUri(uri): OriginalUri,
    RawQuery(raw): RawQuery,
) -> Result<Response> {
    let params = Params::parse(raw.as_deref());
    let mut q = ItemQuery::from_params(&params);
    q.only_trashed = true;
    item_listing(&app, &lib, uri.path(), raw.as_deref(), &params, q).await
}

/// `GET /items/<key>/children`: the item's attachments and notes.
async fn item_children(
    State(app): State<AppState>,
    Extension(lib): Lib,
    Path((_, key)): Path<(String, String)>,
    OriginalUri(uri): OriginalUri,
    RawQuery(raw): RawQuery,
) -> Result<Response> {
    let params = Params::parse(raw.as_deref());
    let mut q = ItemQuery::from_params(&params);
    q.parent = Some(key);
    item_listing(&app, &lib, uri.path(), raw.as_deref(), &params, q).await
}

async fn collection_items(
    State(app): State<AppState>,
    Extension(lib): Lib,
    Path((_, key)): Path<(String, String)>,
    OriginalUri(uri): OriginalUri,
    RawQuery(raw): RawQuery,
) -> Result<Response> {
    let params = Params::parse(raw.as_deref());
    let mut q = ItemQuery::from_params(&params);
    q.collection = Some(key);
    item_listing(&app, &lib, uri.path(), raw.as_deref(), &params, q).await
}

async fn collection_items_top(
    State(app): State<AppState>,
    Extension(lib): Lib,
    Path((_, key)): Path<(String, String)>,
    OriginalUri(uri): OriginalUri,
    RawQuery(raw): RawQuery,
) -> Result<Response> {
    let params = Params::parse(raw.as_deref());
    let mut q = ItemQuery::from_params(&params);
    q.collection = Some(key);
    q.top = true;
    item_listing(&app, &lib, uri.path(), raw.as_deref(), &params, q).await
}

// ── tags, settings, deletions ────────────────────────────────────────────────

async fn tags_get(Extension(lib): Lib) -> Result<Response> {
    let version = current(&lib).await?;
    let tags = lib.db.read(store::tags).await?;
    Ok((version_headers(version), Json(tags)).into_response())
}

/// `DELETE /tags?tag=a || b`. The sync client sends `tags`, the documented
/// API `tag`; both are accepted.
async fn tags_delete(
    Extension(lib): Lib,
    Extension(who): Who,
    headers: HeaderMap,
    RawQuery(raw): RawQuery,
) -> Result<Response> {
    let expected = require_unmodified(&headers)?;
    let params = Params::parse(raw.as_deref());
    let tags: Vec<String> = params
        .get("tags")
        .or_else(|| params.get("tag"))
        .map(|raw| {
            raw.split("||")
                .map(|t| t.trim().to_string())
                .filter(|t| !t.is_empty())
                .collect()
        })
        .unwrap_or_default();
    match lib
        .db
        .write(move |c| store::delete_tags(c, &tags, Some(expected), who.0.as_ref()))
        .await?
    {
        Outcome::Done(version) => {
            Ok((StatusCode::NO_CONTENT, version_headers(version)).into_response())
        }
        Outcome::Conflict(current) => Ok(conflict(current)),
    }
}

/// `GET /settings`. The client sends its cursor as `?since=` or as
/// `If-Modified-Since-Version`; the higher counts, and this is the one read
/// that answers 304.
async fn settings_get(
    Extension(lib): Lib,
    headers: HeaderMap,
    RawQuery(raw): RawQuery,
) -> Result<Response> {
    let since = Params::parse(raw.as_deref())
        .since()
        .max(header_i64(&headers, "if-modified-since-version").unwrap_or(0));
    let version = current(&lib).await?;
    if since > 0 && since >= version {
        return Ok((StatusCode::NOT_MODIFIED, version_headers(version)).into_response());
    }
    let settings = lib.db.read(move |c| store::settings(c, since)).await?;
    Ok((version_headers(version), Json(settings)).into_response())
}

async fn settings_post(Extension(lib): Lib, headers: HeaderMap, body: Bytes) -> Result<Response> {
    let expected = require_unmodified(&headers)?;
    let value: Value = json_body(&body)?;
    match lib
        .db
        .write(move |c| store::write_settings(c, value, Some(expected)))
        .await?
    {
        Outcome::Done(version) => {
            Ok((StatusCode::NO_CONTENT, version_headers(version)).into_response())
        }
        Outcome::Conflict(current) => Ok(conflict(current)),
    }
}

async fn settings_delete(
    Extension(lib): Lib,
    headers: HeaderMap,
    RawQuery(raw): RawQuery,
) -> Result<Response> {
    let expected = require_unmodified(&headers)?;
    let keys = Params::parse(raw.as_deref()).csv("settingKey");
    match lib
        .db
        .write(move |c| store::delete_settings(c, &keys, Some(expected)))
        .await?
    {
        Outcome::Done(version) => {
            Ok((StatusCode::NO_CONTENT, version_headers(version)).into_response())
        }
        Outcome::Conflict(current) => Ok(conflict(current)),
    }
}

async fn deleted(Extension(lib): Lib, RawQuery(raw): RawQuery) -> Result<Response> {
    let since = Params::parse(raw.as_deref()).since();
    let version = current(&lib).await?;
    let deleted = lib.db.read(move |c| store::deleted(c, since)).await?;
    Ok((version_headers(version), Json(deleted)).into_response())
}

// ── full text ────────────────────────────────────────────────────────────────

async fn fulltext_versions(Extension(lib): Lib, RawQuery(raw): RawQuery) -> Result<Response> {
    let since = Params::parse(raw.as_deref()).since();
    let version = current(&lib).await?;
    let map = lib
        .db
        .read(move |c| store::fulltext_versions(c, since))
        .await?;
    Ok((version_headers(version), Json(map)).into_response())
}

/// `GET /items/<key>/fulltext`, with the row's own version, which must equal
/// the item's entry in the versions map or the client re-downloads it forever.
async fn fulltext_item(
    Extension(lib): Lib,
    Path((_, key)): Path<(String, String)>,
) -> Result<Response> {
    match lib.db.read(move |c| store::fulltext_item(c, &key)).await? {
        Some((version, content)) => Ok((version_headers(version), Json(content)).into_response()),
        None => Err(AppError::NotFound("no full text".into())),
    }
}

async fn fulltext_write(Extension(lib): Lib, headers: HeaderMap, body: Bytes) -> Result<Response> {
    let batch: Vec<Value> = json_body(&body)?;
    let expected = require_unmodified(&headers)?;
    match lib
        .db
        .write(move |c| store::write_fulltext(c, batch, Some(expected)))
        .await?
    {
        Outcome::Done(result) => Ok(write_result(result)),
        Outcome::Conflict(current) => Ok(conflict(current)),
    }
}

// ── files ────────────────────────────────────────────────────────────────────

/// `POST /items/<key>/file` serves two steps of the upload protocol:
/// authorization (form `md5`, `filename`, `filesize`, `mtime`) and, after the
/// bytes were sent to the upload URL, registration (form `upload=<token>`).
async fn file_post(
    State(app): State<AppState>,
    Extension(lib): Lib,
    Extension(who): Who,
    Path((_, key)): Path<(String, String)>,
    headers: HeaderMap,
    Form(form): Form<HashMap<String, String>>,
) -> Result<Response> {
    if !store::valid_object_key(&key) {
        return Err(AppError::BadRequest("invalid item key".into()));
    }

    if let Some(token) = form.get("upload") {
        let upload = app
            .uploads
            .get(token)
            .ok_or_else(|| AppError::BadRequest("no pending upload".into()))?;
        if upload.library != lib.id || upload.item_key != key {
            return Err(AppError::BadRequest(
                "upload does not belong to this item".into(),
            ));
        }
        if !upload.uploaded {
            return Err(AppError::BadRequest("the file was not uploaded".into()));
        }
        // Again, now: other uploads authorized meanwhile may have used it up.
        if let Some(refused) = over_quota(&app, &lib, &key, upload.filesize).await? {
            app.uploads.remove(token);
            app.storage.discard(token).await;
            return Ok(refused);
        }
        app.storage.commit(token, &lib.files, &key).await?;
        app.uploads.remove(token);
        let file = store::FileMeta {
            md5: upload.md5.to_lowercase(),
            filename: upload.filename,
            filesize: upload.filesize,
            mtime: upload.mtime,
            content_type: upload.content_type,
            compressed: upload.zip_md5.is_some(),
        };
        let version = lib
            .db
            .write(move |c| store::register_file(c, &key, &file, who.0.as_ref()))
            .await?;
        note_usage(&app, &lib).await;
        return Ok((StatusCode::NO_CONTENT, version_headers(version)).into_response());
    }

    // Authorization: `If-None-Match: *` for a new file, `If-Match: <old md5>`
    // to replace one. Without either the version guard would be bypassed.
    let if_none_match = headers.contains_key(header::IF_NONE_MATCH);
    let if_match = headers
        .get(header::IF_MATCH)
        .and_then(|v| v.to_str().ok())
        .map(|s| s.trim_matches('"').to_string());
    if !if_none_match && if_match.is_none() {
        return Err(AppError::PreconditionRequired(
            "If-Match or If-None-Match required".into(),
        ));
    }
    let md5_field = |name: &str| {
        form.get(name)
            .map(|s| s.trim().to_lowercase())
            .filter(|s| s.len() == 32 && s.bytes().all(|b| b.is_ascii_hexdigit()))
    };
    let md5 = md5_field("md5").ok_or_else(|| AppError::BadRequest("md5 required".into()))?;
    // Snapshots and other multi-file attachments come as a ZIP of the
    // attachment directory: `md5` is still the main file's, `filesize` the ZIP's.
    let zip_md5 = match form.get("zipMD5") {
        None => None,
        Some(_) => Some(
            md5_field("zipMD5")
                .ok_or_else(|| AppError::BadRequest("zipMD5 must be an md5".into()))?,
        ),
    };
    let filesize: i64 = form
        .get("filesize")
        .and_then(|s| s.parse().ok())
        .filter(|n| *n >= 0)
        .ok_or_else(|| AppError::BadRequest("filesize required".into()))?;
    if filesize as u64 > app.cfg.max_upload {
        return Ok((StatusCode::PAYLOAD_TOO_LARGE, "file too large").into_response());
    }

    // As zotero.org's dataserver does: a 412 carries the *item's* version, which
    // Zotero compares with its own copy of the item, and none when `If-Match`
    // names a file the server does not have, so Zotero retries with
    // `If-None-Match: *` (a file added through Zotero's local API arrives with
    // its hash already marked as synced).
    let lookup = key.clone();
    let (item_version, stored_md5) = lib
        .db
        .read(move |c| store::item_file_state(c, &lookup))
        .await?
        .ok_or_else(|| AppError::NotFound("item not found".into()))?;
    match (&if_match, stored_md5.as_deref()) {
        (Some(_), None) => {
            return Ok((
                StatusCode::PRECONDITION_FAILED,
                "If-Match set but file does not exist",
            )
                .into_response());
        }
        (Some(want), Some(stored)) if !stored.eq_ignore_ascii_case(want) => {
            return Ok(conflict(item_version));
        }
        (_, Some(stored)) if stored.eq_ignore_ascii_case(&md5) => {
            return Ok(
                (version_headers(item_version), Json(json!({ "exists": 1 }))).into_response(),
            );
        }
        (None, Some(_)) => return Ok(conflict(item_version)),
        _ => {}
    }

    if let Some(refused) = over_quota(&app, &lib, &key, filesize).await? {
        return Ok(refused);
    }

    let (token, expired) = app.uploads.insert(PendingUpload {
        library: lib.id,
        item_key: key,
        md5,
        filename: form.get("filename").cloned().unwrap_or_default(),
        filesize,
        mtime: form.get("mtime").and_then(|s| s.parse().ok()).unwrap_or(0),
        content_type: form.get("contentType").filter(|s| !s.is_empty()).cloned(),
        zip_md5,
        uploaded: false,
    });
    for (stale, _) in expired {
        app.storage.discard(&stale).await;
    }
    // Empty prefix/suffix: the client sends the raw bytes to `url`.
    Ok(Json(json!({
        "url": format!("{}/uploads/{token}", app.cfg.public_url),
        "uploadKey": token,
        "contentType": "application/octet-stream",
        "prefix": "",
        "suffix": "",
    }))
    .into_response())
}

/// `POST /uploads/<token>`: receive the bytes of an authorized upload, stream
/// them to object storage, and check them against the authorized md5 and size.
async fn upload(
    State(app): State<AppState>,
    Path(token): Path<String>,
    body: Body,
) -> Result<Response> {
    let pending = app
        .uploads
        .get(&token)
        .ok_or_else(|| AppError::NotFound("unknown upload".into()))?;
    let declared = pending.filesize as u64;
    let received = match app
        .storage
        .receive(&token, body.into_data_stream(), declared)
        .await
    {
        Ok(received) => Some(received),
        Err(AppError::TooLarge) => None,
        Err(e) => return Err(e),
    };
    let want_md5 = pending.zip_md5.as_ref().unwrap_or(&pending.md5);
    let matches = received
        .as_ref()
        .is_some_and(|r| r.size == declared && &r.md5 == want_md5);
    if !matches {
        tracing::warn!(
            item = pending.item_key,
            ?received,
            want_md5,
            want_size = declared,
            "uploaded bytes do not match"
        );
        app.storage.discard(&token).await;
        return Err(AppError::BadRequest(
            "uploaded bytes do not match md5/filesize".into(),
        ));
    }
    app.uploads.update(&token, |u| u.uploaded = true);
    Ok(StatusCode::CREATED.into_response())
}

/// Abandon an authorized upload (not part of the Zotero protocol; lets a
/// client clean up early instead of waiting for the token to expire).
async fn upload_cancel(State(app): State<AppState>, Path(token): Path<String>) -> StatusCode {
    if app.uploads.remove(&token).is_some() {
        app.storage.discard(&token).await;
    }
    StatusCode::NO_CONTENT
}

/// `GET /items/<key>/file`: a `302` the client reads without following — it
/// takes md5 and mtime from the headers and then fetches `Location`, a
/// pre-signed bucket URL (or, without one, an expiring download link here).
async fn file_get(
    State(app): State<AppState>,
    Extension(lib): Lib,
    Extension(principal): Extension<auth::Principal>,
    Path((_, key)): Path<(String, String)>,
) -> Result<Response> {
    if !store::valid_object_key(&key) {
        return Err(AppError::NotFound("file not found".into()));
    }
    let lookup = key.clone();
    let file = lib
        .db
        .read(move |c| store::file_meta(c, &lookup))
        .await?
        .ok_or_else(|| AppError::NotFound("file not found".into()))?;
    let location = match app.storage.signed_download(&lib.files, &key).await? {
        Some(url) => url,
        None => {
            let (token, _) = app.downloads.insert(PendingDownload {
                library: lib.id,
                item_key: key,
                user_id: principal.user_id,
            });
            format!("{}/downloads/{token}", app.cfg.public_url)
        }
    };
    let mut headers = HeaderMap::new();
    headers.insert(header::LOCATION, header_value(&location));
    headers.insert("zotero-file-md5", header_value(&file.md5));
    headers.insert(
        "zotero-file-modification-time",
        HeaderValue::from(file.mtime),
    );
    headers.insert(
        "zotero-file-compressed",
        HeaderValue::from_static(if file.compressed { "Yes" } else { "No" }),
    );
    Ok((StatusCode::FOUND, headers).into_response())
}

/// `GET /downloads/<token>`: stream a file whose download was authorized.
async fn download(State(app): State<AppState>, Path(token): Path<String>) -> Result<Response> {
    let PendingDownload {
        library,
        item_key,
        user_id,
    } = app
        .downloads
        .get(&token)
        .ok_or_else(|| AppError::NotFound("unknown or expired download".into()))?;
    // Taken out of the group, or the group deleted, since it was handed out.
    if let LibraryRef::Group(group) = library
        && app.control.group_access(group, user_id).await?.is_none()
    {
        return Err(AppError::NotFound("unknown or expired download".into()));
    }
    let lib = app.libraries.get(library).await?;
    let lookup = item_key.clone();
    let file = lib.db.read(move |c| store::file_meta(c, &lookup)).await?;
    let (body, size) = app.storage.download(&lib.files, &item_key).await?;
    let mut headers = HeaderMap::new();
    let compressed = file.as_ref().is_some_and(|f| f.compressed);
    let content_type = if compressed {
        "application/zip"
    } else {
        file.as_ref()
            .and_then(|f| f.content_type.as_deref())
            .unwrap_or("application/octet-stream")
    };
    headers.insert(header::CONTENT_TYPE, header_value(content_type));
    headers.insert(header::CONTENT_LENGTH, HeaderValue::from(size));
    if let Some(file) = &file {
        let name: String = if compressed {
            format!("{item_key}.zip")
        } else {
            file.filename
                .chars()
                .filter(|c| !c.is_control() && *c != '"' && *c != '\\')
                .collect()
        };
        headers.insert(
            header::CONTENT_DISPOSITION,
            header_value(&format!("attachment; filename=\"{name}\"")),
        );
    }
    Ok((headers, body).into_response())
}

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

    #[test]
    fn next_link_replaces_start() {
        assert_eq!(
            next_link(
                "https://z.example",
                "/api/users/1/items",
                Some("q=a&start=0&limit=2"),
                2
            ),
            "<https://z.example/api/users/1/items?q=a&limit=2&start=2>; rel=\"next\""
        );
    }
}