defter 0.8.0

//! Server-side commonmeta sync: the library of each user who opted in on
//! their dashboard is enhanced with what the commonmeta store knows about its
//! works, as `commonmeta sync` does against Zotero's local API.
//!
//! It runs in defter's own processes: `defter sync`, which commonmeta's
//! crontab starts (`commonmeta jobs`), syncs everyone who opted in; the
//! server runs the previews and syncs users start on their dashboard. A lock
//! file in the data directory keeps the two from running at once. commonmeta's
//! sync reads and writes the library through [`ZoteroLibrary`], which
//! [`ServerLibrary`] implements over the library's database and file storage —
//! the same writes the API makes, so Zotero clients pick the changes up on
//! their next sync. The commonmeta store (`--commonmeta-db`) is only read: it
//! has a single writer, commonmeta, and the sync fetches nothing from Crossref
//! or DataCite to keep in it.

use std::fs::File;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Duration;

use commonmeta::zotero::{
    self, LibraryItem, PatchOutcome, Schema, SyncOptions, SyncResult, SyncSummary, ZoteroLibrary,
};
use serde_json::{Map, Value};

use crate::app::AppState;
use crate::control::{self, MAX_SYNC_RESULTS, RunStatus, SyncRun, SyncSetting, User};
use crate::error::AppError;
use crate::query::{ItemQuery, Params, QMode};
use crate::registry::Library;
use crate::store::{self, Kind, Outcome, Single};

/// The server-side sync's settings, which turn it on.
#[derive(Debug, Clone)]
pub struct SyncConfig {
    /// The commonmeta store, `commonmeta.sqlite3`, opened read-only.
    pub store: PathBuf,
    /// The lock file a run holds, so that `defter sync` and the server never
    /// sync at once (see [`lock_path`]).
    pub lock: PathBuf,
    /// At most this many PDFs attached per library and run.
    pub max_pdfs: usize,
    /// At most this many items changed or created per library and run; the
    /// rest are left for the next.
    pub limit: usize,
}

/// The lock file of the data directory `data_dir`.
pub fn lock_path(data_dir: &Path) -> PathBuf {
    data_dir.join("sync.lock")
}

fn open_lock(path: &Path) -> std::io::Result<File> {
    std::fs::OpenOptions::new()
        .create(true)
        .truncate(false)
        .write(true)
        .open(path)
}

/// The lock at `path` if no other run holds it, in this process or another;
/// `None` when one does.
pub fn try_lock(path: &Path) -> Result<Option<File>, String> {
    let file = open_lock(path).map_err(|e| format!("opening {}: {e}", path.display()))?;
    match file.try_lock() {
        Ok(()) => Ok(Some(file)),
        Err(std::fs::TryLockError::WouldBlock) => Ok(None),
        Err(std::fs::TryLockError::Error(e)) => Err(format!("locking {}: {e}", path.display())),
    }
}

/// The lock file, held: what [`run_all`] needs.
pub struct Locked {
    _file: File,
}

/// The lock at `path`, once the run holding it, if any, has finished; or
/// `defter restore` or `defter migrate`, which hold it while they replace the
/// databases.
pub async fn lock(path: &Path) -> Result<Locked, String> {
    if let Some(file) = try_lock(path)? {
        return Ok(Locked { _file: file });
    }
    tracing::info!(lock = %path.display(), "commonmeta sync: waiting for the run going on");
    let path = path.to_path_buf();
    tokio::task::spawn_blocking(move || {
        let file = open_lock(&path).map_err(|e| format!("opening {}: {e}", path.display()))?;
        file.lock()
            .map_err(|e| format!("locking {}: {e}", path.display()))?;
        Ok(Locked { _file: file })
    })
    .await
    .map_err(|e| format!("lock task: {e}"))?
}

/// What runs now in this process: one run at a time.
#[derive(Default)]
pub struct SyncState {
    lock: Arc<tokio::sync::Mutex<()>>,
    /// The user whose library is being synced.
    running: std::sync::Mutex<Option<u64>>,
}

impl SyncState {
    /// Whether `user_id`'s library is being synced now, by this process.
    pub fn is_running(&self, user_id: u64) -> bool {
        *self.running.lock().unwrap_or_else(|e| e.into_inner()) == Some(user_id)
    }

    fn set_running(&self, user_id: Option<u64>) {
        *self.running.lock().unwrap_or_else(|e| e.into_inner()) = user_id;
    }
}

/// How the runs of [`run_all`] ended.
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub struct Ran {
    /// Runs that ended `ok`.
    pub ok: usize,
    /// Runs that ended `partial` or `aborted`.
    pub partial: usize,
    /// Runs that ended `failed`, or could not be recorded.
    pub failed: usize,
}

/// The users [`run_all`] syncs, with their settings: everyone who opted in,
/// or only `only`. A preview of `only` needs no opting in, as on the dashboard.
async fn jobs(
    app: &AppState,
    only: Option<u64>,
    dry_run: bool,
) -> Result<Vec<(User, SyncSetting)>, AppError> {
    let mut jobs = app.control.sync_jobs(app.cfg.user_id).await?;
    let Some(id) = only else {
        return Ok(jobs);
    };
    jobs.retain(|(user, _)| user.id == id);
    if jobs.is_empty()
        && dry_run
        && let Some(user) = app.control.user(id).await?.filter(|u| !u.disabled)
    {
        let setting = app.control.sync_setting(id).await?;
        jobs.push((user, setting));
    }
    if jobs.is_empty() {
        return Err(AppError::NotFound(format!(
            "user {id} does not exist, is disabled or has not turned the sync on"
        )));
    }
    Ok(jobs)
}

/// Sync the library of everyone who opted in, the owner first, then whoever
/// waited longest, or only `only`'s; a preview with `dry_run`. The caller
/// holds the lock file ([`lock`]) from before it opens the databases, so a
/// restore cannot replace them under it. A failure is recorded as the user's
/// run and the next goes on; the error is only that the users could not be
/// read.
pub async fn run_all(
    app: &AppState,
    cfg: &SyncConfig,
    _locked: &Locked,
    only: Option<u64>,
    dry_run: bool,
) -> Result<Ran, AppError> {
    let _running = app.sync.lock.lock().await;
    let jobs = jobs(app, only, dry_run).await?;
    tracing::info!(libraries = jobs.len(), dry_run, "commonmeta sync starting");
    let mut ran = Ran::default();
    for (user, setting) in jobs {
        match run_user(app, cfg, &user, setting, dry_run).await {
            Ok(run) if run.status == RunStatus::Ok => ran.ok += 1,
            Ok(run) if run.status == RunStatus::Failed => ran.failed += 1,
            Ok(_) => ran.partial += 1,
            Err(error) => {
                ran.failed += 1;
                tracing::error!(%error, user = user.id, "commonmeta sync: recording the run");
            }
        }
    }
    tracing::info!(
        ok = ran.ok,
        partial = ran.partial,
        failed = ran.failed,
        "commonmeta sync finished"
    );
    Ok(ran)
}

/// Start a run of `user`'s library with `setting` now, in the background,
/// unless a run is going on, here or in `defter sync`: `false` then. A real
/// run writes only while the user is opted in; a preview (`dry_run`) writes
/// nothing.
pub fn start(
    app: &AppState,
    cfg: &SyncConfig,
    user: User,
    setting: SyncSetting,
    dry_run: bool,
) -> bool {
    let Ok(running) = app.sync.lock.clone().try_lock_owned() else {
        return false;
    };
    let locked = match try_lock(&cfg.lock) {
        Ok(Some(file)) => file,
        Ok(None) => return false,
        Err(error) => {
            tracing::error!(%error, "commonmeta sync: the lock file");
            return false;
        }
    };
    let (app, cfg) = (app.clone(), cfg.clone());
    tokio::spawn(async move {
        let _running = (running, locked);
        if let Err(error) = run_user(&app, &cfg, &user, setting, dry_run).await {
            tracing::error!(%error, user = user.id, "commonmeta sync: recording the run");
        }
    });
    true
}

/// Sync `user`'s own library with `setting` and record the run. The error is
/// only that of recording it; a failed sync is a run with status `failed`.
pub async fn run_user(
    app: &AppState,
    cfg: &SyncConfig,
    user: &User,
    setting: SyncSetting,
    dry_run: bool,
) -> Result<SyncRun, AppError> {
    let started_at = control::now();
    app.sync.set_running(Some(user.id));
    let outcome = sync_library(app, cfg, user, setting, dry_run).await;
    app.sync.set_running(None);
    let (status, summary, results, error) = match outcome {
        Ok((results, summary, aborted)) => {
            let status = if aborted {
                RunStatus::Aborted
            } else if summary.failed > 0 || summary.conflicts > 0 {
                RunStatus::Partial
            } else {
                RunStatus::Ok
            };
            let error = aborted.then(|| "sync was turned off while it ran".to_string());
            (status, summary, results, error)
        }
        Err(error) => (
            RunStatus::Failed,
            SyncSummary::default(),
            Vec::new(),
            Some(error),
        ),
    };
    let results_total = results.len() as i64;
    let kept = kept(&results, MAX_SYNC_RESULTS);
    let run = SyncRun {
        user_id: user.id,
        started_at,
        finished_at: control::now(),
        status,
        dry_run,
        commonmeta_version: commonmeta::VERSION.to_string(),
        summary: serde_json::to_string(&summary).unwrap_or_else(|_| "{}".into()),
        error,
        results: Some(serde_json::to_string(&kept).unwrap_or_else(|_| "[]".into())),
        results_total,
    };
    tracing::info!(
        user = user.id,
        status = status.as_str(),
        dry_run,
        items = summary.items,
        updated = summary.updated,
        created = summary.created,
        pdfs = summary.pdfs,
        failed = summary.failed,
        conflicts = summary.conflicts,
        error = run.error.as_deref(),
        seconds = run.finished_at - started_at,
        "commonmeta sync of a library"
    );
    app.control.record_run(run.clone()).await?;
    Ok(run)
}

/// At most `max` of `results`, in their order: those with changes (their old
/// and new values) first, then those with fields given, then the rest. A run
/// can have more than it changes: PDFs, merges, conflicts and failures.
fn kept(results: &[SyncResult], max: usize) -> Vec<&SyncResult> {
    let mut order: Vec<usize> = (0..results.len()).collect();
    if results.len() > max {
        order.sort_by_key(|&i| {
            let r = &results[i];
            (r.changes.is_empty(), r.fields.is_empty())
        });
        order.truncate(max);
        order.sort_unstable();
    }
    order.into_iter().map(|i| &results[i]).collect()
}

/// Run commonmeta's sync over `user`'s library on the blocking pool: its
/// results, counts and whether it was turned off meanwhile, or why it failed.
async fn sync_library(
    app: &AppState,
    cfg: &SyncConfig,
    user: &User,
    setting: SyncSetting,
    dry_run: bool,
) -> Result<(Vec<SyncResult>, SyncSummary, bool), String> {
    let library = app.user_library(user).await.map_err(|e| e.to_string())?;
    let schema = crate::zotero::schema_json(app)
        .await
        .map_err(|e| e.to_string())?;
    let schema: Value = serde_json::from_str(&schema).map_err(|e| format!("Zotero schema: {e}"))?;
    let opts = SyncOptions {
        library: format!("users/{}", user.id),
        // PDFs come with the works' metadata.
        pdf: setting.metadata,
        max_pdfs: cfg.max_pdfs,
        dry_run,
        limit: cfg.limit,
        // Only what the store holds: what is fetched would be kept in it.
        network: false,
        // Items sharing a DOI are left for Zotero to merge, versions of a work
        // as they are.
        versions: false,
        // Another type is a difference too: a duplicate of the work's type is
        // made, for Zotero to merge.
        types: true,
        metadata: setting.metadata,
        dois: Vec::new(),
        item_keys: Vec::new(),
        orcid: user.orcid.clone().filter(|_| setting.publications),
    };
    let mut server = ServerLibrary {
        app: app.clone(),
        user_id: user.id,
        library,
        schema,
        runtime: tokio::runtime::Handle::current(),
        aborted: false,
        public_only: true,
    };
    let store = cfg.store.clone();
    tokio::task::spawn_blocking(move || {
        let synced = zotero::sync(&store, &mut server, &opts);
        match synced {
            Ok((results, summary)) => Ok((results, summary, server.aborted)),
            // Turned off meanwhile, which stopped it: aborted, not failed.
            Err(_) if server.aborted => Ok((Vec::new(), SyncSummary::default(), true)),
            Err(e) => Err(e.to_string()),
        }
    })
    .await
    .map_err(|e| format!("sync task: {e}"))?
}

/// A user's library in this process, as commonmeta's sync reads and writes
/// it. Its methods run on the blocking pool and wait there for the database.
struct ServerLibrary {
    app: AppState,
    user_id: u64,
    library: Arc<Library>,
    schema: Value,
    runtime: tokio::runtime::Handle,
    /// The user turned sync off, or was disabled, during the run.
    aborted: bool,
    /// Download PDFs only from public addresses (off in tests, which serve
    /// them from loopback).
    public_only: bool,
}

type CmResult<T> = commonmeta::Result<T>;

fn failed(e: impl std::fmt::Display) -> commonmeta::Error {
    commonmeta::Error::Http(format!("defter: {e}"))
}

/// The objects of a listing as commonmeta reads them.
fn library_items(objects: Vec<Value>) -> Vec<LibraryItem> {
    objects.iter().filter_map(LibraryItem::from_value).collect()
}

/// A regular item: not an attachment, note or annotation, and not in the trash.
fn is_regular(item: &LibraryItem) -> bool {
    let item_type = item
        .data
        .get("itemType")
        .and_then(Value::as_str)
        .unwrap_or("");
    let deleted = item
        .data
        .get("deleted")
        .is_some_and(|d| d.as_bool().unwrap_or(d == 1));
    !matches!(item_type, "attachment" | "note" | "annotation" | "") && !deleted
}

impl ServerLibrary {
    fn wait<T>(&self, future: impl std::future::Future<Output = T>) -> T {
        self.runtime.block_on(future)
    }

    /// Items matching `q`, every page.
    fn query(&self, mut q: ItemQuery) -> CmResult<Vec<LibraryItem>> {
        q.limit = i64::MAX;
        q.start = 0;
        let db = self.library.db.clone();
        let (objects, _) = self
            .wait(db.read(move |c| store::query_items(c, &q)))
            .map_err(failed)?;
        Ok(library_items(objects))
    }

    /// Fails once the user turned sync off or was disabled, so that nothing
    /// more is written to their library.
    fn may_write(&mut self) -> CmResult<()> {
        if !self.aborted {
            let access = self
                .wait(self.app.control.sync_access(self.user_id))
                .map_err(failed)?;
            self.aborted = access.is_none();
        }
        if self.aborted {
            return Err(failed("sync was turned off"));
        }
        Ok(())
    }

    fn write_one(
        &mut self,
        key: &str,
        version: i64,
        data: &Map<String, Value>,
    ) -> CmResult<PatchOutcome> {
        self.may_write()?;
        let (db, key, data) = (
            self.library.db.clone(),
            key.to_string(),
            Value::Object(data.clone()),
        );
        let outcome = self
            .wait(
                db.write(move |c| store::write_one(c, Kind::Item, &key, data, version, true, None)),
            )
            .map_err(failed)?;
        match outcome {
            Single::Done { .. } => Ok(PatchOutcome::Updated),
            Single::Conflict(_) => Ok(PatchOutcome::Conflict),
            Single::NotFound => Err(failed("item not found")),
            Single::Invalid(message) => Err(failed(message)),
        }
    }

    /// Create `items`: each one's key and version, or why it was refused.
    fn create(&mut self, items: Vec<Value>) -> CmResult<Vec<Result<(String, i64), String>>> {
        if items.is_empty() {
            return Ok(Vec::new());
        }
        // Refused one by one, so the sync goes on to report what it did.
        if let Err(refused) = self.may_write() {
            return Ok(vec![Err(refused.to_string()); items.len()]);
        }
        let count = items.len();
        let db = self.library.db.clone();
        let outcome = self
            .wait(db.write(move |c| store::write_batch(c, Kind::Item, items, None, None)))
            .map_err(failed)?;
        let Outcome::Done(result) = outcome else {
            return Err(failed("the library changed meanwhile"));
        };
        Ok((0..count)
            .map(|i| {
                let i = i.to_string();
                match result.successful.get(&i) {
                    Some(done) => Ok((
                        done["key"].as_str().unwrap_or_default().to_string(),
                        done["version"].as_i64().unwrap_or(result.version),
                    )),
                    None => Err(result
                        .failed
                        .get(&i)
                        .and_then(|f| f["message"].as_str())
                        .unwrap_or("no item created")
                        .to_string()),
                }
            })
            .collect())
    }

    /// Store `bytes` as the PDF file of attachment `key`, as an upload would.
    fn store_pdf(&self, key: &str, bytes: Vec<u8>, filename: String) -> Result<(), AppError> {
        let (app, library) = (self.app.clone(), self.library.clone());
        let key = key.to_string();
        self.wait(async move {
            let size = bytes.len() as i64;
            if size as u64 > app.cfg.max_upload {
                return Err(AppError::TooLarge);
            }
            if crate::zotero::over_quota(&app, &library, &key, size)
                .await?
                .is_some()
            {
                return Err(AppError::BadRequest(
                    "the PDF would exceed the quota".into(),
                ));
            }
            let md5 = format!("{:x}", md5::compute(&bytes));
            let token = crate::auth::random_hex(16);
            let body = futures::stream::iter([Ok::<_, std::convert::Infallible>(
                bytes::Bytes::from(bytes),
            )]);
            app.storage.receive(&token, body, size as u64).await?;
            if let Err(error) = app.storage.commit(&token, &library.files, &key).await {
                app.storage.discard(&token).await;
                return Err(error);
            }
            let file = store::FileMeta {
                md5,
                filename,
                filesize: size,
                mtime: control::now() * 1000,
                content_type: Some("application/pdf".into()),
                compressed: false,
            };
            library
                .db
                .write(move |c| store::register_file(c, &key, &file, None))
                .await?;
            crate::zotero::note_usage(&app, &library).await;
            Ok(())
        })
    }
}

/// The largest PDF downloaded.
const PDF_MAX_BYTES: u64 = 100 * 1024 * 1024;

/// Redirects followed for a PDF.
const PDF_MAX_REDIRECTS: usize = 5;

/// Whether `ip` is an address on the internet: not loopback, private,
/// link-local, shared (CGNAT), multicast, documentation or unspecified. A PDF
/// URL comes from a work's metadata, which its depositor wrote, so the server
/// must not be sent to its own or its network's services.
fn is_public(ip: std::net::IpAddr) -> bool {
    use std::net::IpAddr;
    match ip {
        IpAddr::V4(a) => {
            let [first, second, ..] = a.octets();
            !(a.is_loopback()
                || a.is_private()
                || a.is_link_local()
                || a.is_broadcast()
                || a.is_documentation()
                || a.is_unspecified()
                || first == 0
                || first >= 224
                || (first == 100 && (second & 0xc0) == 64))
        }
        IpAddr::V6(a) => match a.to_ipv4_mapped() {
            Some(v4) => is_public(IpAddr::V4(v4)),
            None => {
                let first = a.segments()[0];
                !(a.is_loopback()
                    || a.is_unspecified()
                    || a.is_multicast()
                    || (first & 0xfe00) == 0xfc00
                    || (first & 0xffc0) == 0xfe80
                    || first == 0x2001 && a.segments()[1] == 0x0db8)
            }
        },
    }
}

/// Download the PDF at `url`: at most `max` bytes, and with `public_only`
/// only from public addresses, checked for every redirect and pinned for the
/// request, so a name cannot resolve elsewhere in between. Anything that is
/// not a PDF (a landing page, say) is refused.
async fn fetch_pdf(url: &str, max: u64, public_only: bool) -> Result<Vec<u8>, String> {
    let mut url = reqwest::Url::parse(url).map_err(|e| format!("{url}: {e}"))?;
    for _ in 0..=PDF_MAX_REDIRECTS {
        if !matches!(url.scheme(), "http" | "https") {
            return Err(format!("{url}: not an http(s) URL"));
        }
        let host = url
            .host_str()
            .ok_or_else(|| format!("{url}: no host"))?
            .to_string();
        let port = url.port_or_known_default().unwrap_or(443);
        let lookup = host
            .trim_start_matches('[')
            .trim_end_matches(']')
            .to_string();
        let addrs: Vec<std::net::SocketAddr> = tokio::net::lookup_host((lookup.as_str(), port))
            .await
            .map_err(|e| format!("{host}: {e}"))?
            .collect();
        let Some(addr) = addrs.first().copied() else {
            return Err(format!("{host}: no address"));
        };
        if public_only && addrs.iter().any(|a| !is_public(a.ip())) {
            return Err(format!("{host}: not a public address"));
        }
        let client = reqwest::Client::builder()
            .user_agent(concat!("defter/", env!("CARGO_PKG_VERSION")))
            .redirect(reqwest::redirect::Policy::none())
            .timeout(Duration::from_secs(120))
            .resolve(&host, addr)
            .build()
            .map_err(|e| e.to_string())?;
        let mut resp = client
            .get(url.clone())
            .send()
            .await
            .map_err(|e| format!("{host}: {e}"))?;
        if resp.status().is_redirection() {
            let location = resp
                .headers()
                .get(reqwest::header::LOCATION)
                .and_then(|l| l.to_str().ok())
                .ok_or_else(|| format!("{host}: a redirect without a location"))?;
            url = url.join(location).map_err(|e| format!("{host}: {e}"))?;
            continue;
        }
        if !resp.status().is_success() {
            return Err(format!("{host}: HTTP {}", resp.status().as_u16()));
        }
        let too_large = || format!("{host}: larger than {max} bytes");
        if resp.content_length().is_some_and(|n| n > max) {
            return Err(too_large());
        }
        let mut bytes = Vec::new();
        while let Some(chunk) = resp.chunk().await.map_err(|e| format!("{host}: {e}"))? {
            if (bytes.len() + chunk.len()) as u64 > max {
                return Err(too_large());
            }
            bytes.extend_from_slice(&chunk);
        }
        if !bytes.starts_with(b"%PDF-") {
            return Err(format!("{host}: not a PDF"));
        }
        return Ok(bytes);
    }
    Err(format!("{url}: too many redirects"))
}

impl ZoteroLibrary for ServerLibrary {
    fn schema(&self) -> CmResult<Schema> {
        Schema::from_value(&self.schema)
    }

    fn top_items(&self, _: &str) -> CmResult<Vec<LibraryItem>> {
        let mut q = ItemQuery::from_params(&Params::parse(None));
        q.top = true;
        self.query(q)
    }

    fn children(&self, _: &str, key: &str) -> CmResult<Vec<LibraryItem>> {
        let mut q = ItemQuery::from_params(&Params::parse(None));
        q.parent = Some(key.to_string());
        self.query(q)
    }

    fn scoped_items(
        &self,
        library: &str,
        dois: &[String],
        keys: &[String],
    ) -> CmResult<(Vec<LibraryItem>, Vec<String>)> {
        let mut found: Vec<LibraryItem> = Vec::new();
        let mut missing = Vec::new();
        let db = self.library.db.clone();
        let wanted = keys.to_vec();
        let objects = self
            .wait(db.read(move |c| store::objects(c, Kind::Item, &wanted)))
            .map_err(failed)?;
        let items = library_items(objects);
        for key in keys {
            match items.iter().find(|i| &i.key == key && is_regular(i)) {
                Some(item) => found.push(item.clone()),
                None => missing.push(key.clone()),
            }
        }
        for doi in dois {
            // As the local API is asked: by the DOI's suffix, then exactly.
            let mut q = ItemQuery::from_params(&Params::parse(None));
            q.q = Some(
                doi.split_once('/')
                    .map_or(doi.as_str(), |(_, s)| s)
                    .to_string(),
            );
            q.qmode = QMode::Everything;
            let matches: Vec<LibraryItem> = self
                .query(q)?
                .into_iter()
                .filter(|i| is_regular(i) && i.doi().as_deref() == Some(doi.as_str()))
                .collect();
            if matches.is_empty() {
                missing.push(doi.clone());
            }
            found.extend(matches);
        }
        let mut seen = std::collections::HashSet::new();
        found.retain(|i| seen.insert(i.key.clone()));
        let mut children = Vec::new();
        for item in &found {
            children.extend(self.children(library, &item.key)?);
        }
        found.extend(children.into_iter().filter(|c| seen.insert(c.key.clone())));
        Ok((found, missing))
    }

    fn patch_item(
        &mut self,
        _: &str,
        key: &str,
        version: i64,
        changes: &Map<String, Value>,
    ) -> CmResult<PatchOutcome> {
        self.write_one(key, version, changes)
    }

    fn create_items(&mut self, _: &str, items: &[Value]) -> CmResult<Vec<Result<String, String>>> {
        Ok(self
            .create(items.to_vec())?
            .into_iter()
            .map(|created| created.map(|(key, _)| key))
            .collect())
    }

    fn attach_pdf(&mut self, _: &str, parent: &str, doi: &str, pdf_url: &str) -> CmResult<()> {
        self.may_write()?;
        let max = PDF_MAX_BYTES.min(self.app.cfg.max_upload);
        let bytes = self
            .wait(fetch_pdf(pdf_url, max, self.public_only))
            .map_err(failed)?;
        let (attachment, filename) = zotero::pdf_attachment(parent, doi, pdf_url);
        let (key, version) = match self.create(vec![attachment])?.pop() {
            Some(Ok(created)) => created,
            Some(Err(message)) => return Err(failed(message)),
            None => return Err(failed("no attachment created")),
        };
        if let Err(error) = self.store_pdf(&key, bytes, filename) {
            let (db, item) = (self.library.db.clone(), key.clone());
            let removed =
                self.wait(db.write(move |c| store::delete_one(c, Kind::Item, &item, version)));
            if let Err(e) = removed {
                tracing::warn!(error = %e, key, "removing an attachment without its PDF");
            }
            return Err(failed(error));
        }
        Ok(())
    }
}

impl SyncState {
    /// Hold the run lock, as a run does, for tests.
    #[cfg(test)]
    pub(crate) async fn hold_for_test(&self) -> tokio::sync::OwnedMutexGuard<()> {
        self.lock.clone().lock_owned().await
    }
}

/// `user`'s library as the sync sees it, for tests. Use it on the blocking pool.
#[cfg(test)]
pub(crate) async fn library_for_test(app: &AppState, user: &User) -> Box<dyn ZoteroLibrary + Send> {
    let schema = crate::zotero::schema_json(app).await.unwrap();
    Box::new(ServerLibrary {
        app: app.clone(),
        user_id: user.id,
        library: app.user_library(user).await.unwrap(),
        schema: serde_json::from_str(&schema).unwrap(),
        runtime: tokio::runtime::Handle::current(),
        aborted: false,
        public_only: false,
    })
}

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

    #[test]
    fn results_with_changes_are_kept_first() {
        let result = |key: &str, fields: &[&str], change: bool| SyncResult {
            key: key.into(),
            fields: fields.iter().map(|f| f.to_string()).collect(),
            changes: if change {
                Map::from_iter([(
                    "title".to_string(),
                    serde_json::json!({"from": "", "to": "T"}),
                )])
            } else {
                Map::new()
            },
            ..Default::default()
        };
        let results = [
            result("PDF1", &[], false),
            result("FILL", &["title"], false),
            result("OVER", &["title"], true),
            result("PDF2", &[], false),
        ];
        let keys =
            |max| -> Vec<&str> { kept(&results, max).iter().map(|r| r.key.as_str()).collect() };
        assert_eq!(keys(4), ["PDF1", "FILL", "OVER", "PDF2"]);
        assert_eq!(keys(3), ["PDF1", "FILL", "OVER"]);
        assert_eq!(keys(2), ["FILL", "OVER"]);
        assert_eq!(keys(1), ["OVER"]);
        assert!(keys(0).is_empty());
    }

    #[test]
    fn only_public_addresses_are_fetched_from() {
        for public in [
            "93.184.216.34",
            "2606:2800:220:1::1",
            "::ffff:93.184.216.34",
        ] {
            assert!(is_public(public.parse().unwrap()), "{public}");
        }
        for internal in [
            "127.0.0.1",
            "10.1.2.3",
            "172.16.0.1",
            "192.168.1.1",
            "169.254.169.254",
            "100.64.0.1",
            "0.0.0.0",
            "224.0.0.1",
            "::1",
            "fd00::1",
            "fe80::1",
            "::ffff:127.0.0.1",
        ] {
            assert!(!is_public(internal.parse().unwrap()), "{internal}");
        }
    }

    /// Serve `body` once, without a length, after redirecting from `/old`.
    fn serve_once(body: Vec<u8>) -> String {
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let base = format!("http://{}", listener.local_addr().unwrap());
        std::thread::spawn(move || {
            use std::io::{Read, Write};
            for response in [
                b"HTTP/1.1 302 Found\r\nLocation: /paper.pdf\r\nContent-Length: 0\r\n\r\n".to_vec(),
                [
                    b"HTTP/1.1 200 OK\r\nConnection: close\r\n\r\n".to_vec(),
                    body,
                ]
                .concat(),
            ] {
                let (mut stream, _) = listener.accept().unwrap();
                let _ = stream.read(&mut [0u8; 4096]);
                let _ = stream.write_all(&response);
            }
        });
        format!("{base}/old")
    }

    #[tokio::test]
    async fn a_pdf_is_followed_and_capped() {
        let pdf = [b"%PDF-1.4\n".as_slice(), &[b'x'; 100]].concat();
        let url = serve_once(pdf.clone());
        assert_eq!(fetch_pdf(&url, 1024, false).await.unwrap(), pdf);
        let url = serve_once([b"%PDF-1.4\n".as_slice(), &[b'x'; 4096]].concat());
        let refused = fetch_pdf(&url, 1024, false).await.unwrap_err();
        assert!(refused.contains("larger than 1024 bytes"), "{refused}");
        let url = serve_once(b"<html>a landing page</html>".to_vec());
        let refused = fetch_pdf(&url, 1024, false).await.unwrap_err();
        assert!(refused.contains("not a PDF"), "{refused}");
    }

    #[tokio::test]
    async fn a_pdf_on_an_internal_address_is_refused() {
        let refused = fetch_pdf("http://127.0.0.1:9/paper.pdf", 1024, true)
            .await
            .unwrap_err();
        assert!(refused.contains("not a public address"), "{refused}");
        let refused = fetch_pdf("file:///etc/passwd", 1024, true)
            .await
            .unwrap_err();
        assert!(refused.contains("not an http(s) URL"), "{refused}");
    }
}