defter 0.7.2

//! The control database (`control.sqlite3`): the server's users, and what is
//! known about each library outside it. Each library's data is a database of its
//! own (see `registry`). The schema is `migrations/control`.

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

use rusqlite::{Connection, OptionalExtension, params};

use crate::db::{self, Db};
use crate::error::AppError;
use crate::storage::FileScope;

/// A library: a user's own, or a group's.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum LibraryRef {
    User(u64),
    Group(u64),
}

impl LibraryRef {
    /// `u<id>` or `g<id>`: the name of its database file and its backups.
    pub fn parse(name: &str) -> Option<Self> {
        let (kind, id) = name.split_at_checked(1)?;
        let id = id.parse().ok().filter(|_| !id.starts_with('+'))?;
        match kind {
            "u" => Some(LibraryRef::User(id)),
            "g" => Some(LibraryRef::Group(id)),
            _ => None,
        }
    }
}

impl std::fmt::Display for LibraryRef {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            LibraryRef::User(id) => write!(f, "u{id}"),
            LibraryRef::Group(id) => write!(f, "g{id}"),
        }
    }
}

/// A user of the server.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct User {
    /// Their zotero.org user ID, which their Zotero knows them by.
    pub id: u64,
    pub orcid: Option<String>,
    pub username: String,
    pub display_name: Option<String>,
    pub profile_hidden: bool,
    pub disabled: bool,
    /// Their files are under `files/`, from before there were several libraries.
    pub legacy_files: bool,
    /// Their storage quota in bytes; `None` for the server's default.
    pub quota_bytes: Option<i64>,
    /// Disabled because the provider no longer granted them access, not by an
    /// admin: signing in with it granted again enables them.
    pub access_disabled: bool,
    /// An admin because the provider said so at their last sign-in through it
    /// (`--oidc-admin-values`).
    pub provider_admin: bool,
}

impl User {
    /// What to call them: their name from ORCID, else their user name.
    pub fn name(&self) -> &str {
        self.display_name.as_deref().unwrap_or(&self.username)
    }

    /// Where their library's attachment files are.
    pub fn files(&self) -> FileScope {
        if self.legacy_files {
            FileScope::Legacy
        } else {
            FileScope::User(self.id)
        }
    }
}

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

/// The control database at `path` (created if needed) as one plain connection,
/// for commands that run without the server, such as `defter migrate`.
pub fn open_connection(path: &Path) -> rusqlite::Result<Connection> {
    let mut conn = Connection::open(path)?;
    conn.busy_timeout(std::time::Duration::from_secs(30))?;
    conn.execute_batch("PRAGMA journal_mode = WAL; PRAGMA foreign_keys = ON;")?;
    db::migrate(&mut conn, db::CONTROL)?;
    Ok(conn)
}

const USER_COLUMNS: &str = "id, orcid, username, display_name, profile_hidden, disabled, \
                            files_layout, quota_bytes, access_disabled, provider_admin";

/// How many columns `USER_COLUMNS` are: what a query selects after them starts
/// here.
const USER_COLUMN_COUNT: usize = 10;

fn user_row(r: &rusqlite::Row<'_>) -> rusqlite::Result<User> {
    Ok(User {
        id: r.get::<_, i64>(0)? as u64,
        orcid: r.get(1)?,
        username: r.get(2)?,
        display_name: r.get(3)?,
        profile_hidden: r.get(4)?,
        disabled: r.get(5)?,
        legacy_files: r.get::<_, String>(6)? == "legacy",
        quota_bytes: r.get(7)?,
        access_disabled: r.get(8)?,
        provider_admin: r.get(9)?,
    })
}

pub fn user(conn: &Connection, id: u64) -> rusqlite::Result<Option<User>> {
    conn.query_row(
        &format!("SELECT {USER_COLUMNS} FROM user WHERE id = ?1"),
        [id as i64],
        user_row,
    )
    .optional()
}

pub fn user_by_orcid(conn: &Connection, orcid: &str) -> rusqlite::Result<Option<User>> {
    conn.query_row(
        &format!("SELECT {USER_COLUMNS} FROM user WHERE orcid = ?1"),
        [orcid],
        user_row,
    )
    .optional()
}

pub fn insert_user(conn: &Connection, user: &User) -> rusqlite::Result<()> {
    conn.execute(
        "INSERT INTO user (id, orcid, username, display_name, profile_hidden, disabled,
                           files_layout, quota_bytes, access_disabled, provider_admin,
                           created_at)
         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)",
        params![
            user.id as i64,
            user.orcid,
            user.username,
            user.display_name,
            user.profile_hidden,
            user.disabled,
            if user.legacy_files { "legacy" } else { "user" },
            user.quota_bytes,
            user.access_disabled,
            user.provider_admin,
            now(),
        ],
    )?;
    Ok(())
}

/// Whether a user other than `id` has the user name `username`.
fn name_taken(conn: &Connection, username: &str, id: u64) -> rusqlite::Result<bool> {
    conn.query_row(
        "SELECT count(*) > 0 FROM user WHERE username = ?1 AND id != ?2",
        params![username, id as i64],
        |r| r.get(0),
    )
}

/// Whether `orcid` has an invitation not used yet.
fn invited(conn: &Connection, orcid: &str) -> rusqlite::Result<bool> {
    conn.query_row(
        "SELECT count(*) > 0 FROM invite WHERE orcid = ?1 AND used_by IS NULL",
        [orcid],
        |r| r.get(0),
    )
}

/// Disable user `id` and revoke their keys: by an admin, or, with `access`,
/// because the provider no longer grants them access, which also ends what it
/// made them.
fn disable(conn: &Connection, id: u64, access: bool) -> rusqlite::Result<()> {
    conn.execute(
        "UPDATE user SET disabled = 1, access_disabled = ?2,
                         provider_admin = provider_admin AND NOT ?2
         WHERE id = ?1",
        params![id as i64, access],
    )?;
    conn.execute(
        "UPDATE api_key SET revoked_at = ?2 WHERE user_id = ?1 AND revoked_at IS NULL",
        params![id as i64, now()],
    )?;
    Ok(())
}

/// The sha256 of an API key, hex: all that is kept of a key handed out.
pub fn key_hash(key: &str) -> String {
    aws_lc_rs::digest::digest(&aws_lc_rs::digest::SHA256, key.as_bytes())
        .as_ref()
        .iter()
        .map(|b| format!("{b:02x}"))
        .collect()
}

/// How long a key's last use goes unrecorded: often enough to tell a device in
/// use from one long gone, without a write per request.
const KEY_USE_PRECISION: i64 = 3600;

/// An invitation to sign in and get a library.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Invite {
    pub orcid: String,
    pub note: Option<String>,
    pub created_at: i64,
    /// Who it became, once used.
    pub used_by: Option<u64>,
}

/// An API key handed out to a user's device.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ApiKey {
    pub id: i64,
    /// Its first characters, to tell it apart.
    pub prefix: String,
    pub name: String,
    pub write: bool,
    pub created_at: i64,
    pub last_used_at: Option<i64>,
}

/// A minted key a request presented: whose it is and what it may do.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct KeyOwner {
    pub key_id: i64,
    pub user_id: u64,
    pub write: bool,
}

/// What the provider's group or role claim grants someone signing in through
/// it, when `--oidc-user-values` decides who may use the server.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct Grant {
    /// They carry a user value, or an admin value.
    pub user: bool,
    /// They carry an admin value.
    pub admin: bool,
    /// The provider sent no such claim at all: a mapper missing or detached
    /// there, as likely as someone in no group. They are refused, but nobody's
    /// account changes for it.
    pub missing: bool,
}

/// What signing in with an ORCID iD, or joining, came to.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum SignIn {
    User(User),
    /// Known, but an admin disabled them.
    Disabled,
    /// Neither a user, nor invited, nor an admin.
    NotInvited,
    /// The provider does not grant them access. `disabled` is the user this
    /// sign-in disabled for it, whose sessions end.
    NoAccess {
        disabled: Option<u64>,
    },
    /// Invited or an admin, but no user yet: they join with their zotero.org
    /// account (see `Control::join`).
    NoAccount,
    /// Joining: that zotero.org user ID is another user's here.
    IdTaken,
}

/// A user, with what an admin sees of them.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct UserOverview {
    pub user: User,
    /// The bytes of their attachment files, in their own library and the groups
    /// they own, as last counted.
    pub bytes_used: i64,
    /// Their devices' keys that are not revoked.
    pub keys: i64,
}

/// What a user chose for the server-side commonmeta sync of their library.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct SyncSetting {
    /// Keep the library up to date, daily.
    pub enabled: bool,
    /// Attach the PDFs the store knows to items that have none.
    pub pdf: bool,
    /// Put the works of their ORCID iD into My Publications.
    pub publications: bool,
    /// First merge the items sharing a DOI.
    pub deduplicate: bool,
    /// Let commonmeta's container title, date, authors, license and DOI form
    /// replace the item's, not only fill what it lacks.
    pub overwrite: bool,
}

/// How a sync run ended.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RunStatus {
    /// Everything went through.
    Ok,
    /// Some items failed, or changed meanwhile and were left for the next run.
    Partial,
    /// The sync stopped with an error.
    Failed,
    /// The user turned sync off, or was disabled, while it ran.
    Aborted,
}

impl RunStatus {
    pub fn as_str(&self) -> &'static str {
        match self {
            RunStatus::Ok => "ok",
            RunStatus::Partial => "partial",
            RunStatus::Failed => "failed",
            RunStatus::Aborted => "aborted",
        }
    }

    pub fn parse(text: &str) -> Option<Self> {
        match text {
            "ok" => Some(RunStatus::Ok),
            "partial" => Some(RunStatus::Partial),
            "failed" => Some(RunStatus::Failed),
            "aborted" => Some(RunStatus::Aborted),
            _ => None,
        }
    }
}

/// One sync run of a user's library.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SyncRun {
    pub user_id: u64,
    pub started_at: i64,
    pub finished_at: i64,
    pub status: RunStatus,
    pub dry_run: bool,
    pub commonmeta_version: String,
    /// commonmeta's `SyncSummary`, as JSON.
    pub summary: String,
    pub error: Option<String>,
    /// The per-item results as a JSON array, each change as `{from, to}` in a
    /// preview or with `overwrite`, at most [`MAX_SYNC_RESULTS`]; only the newest
    /// run of a user keeps them.
    pub results: Option<String>,
    /// How many results there were before they were capped.
    pub results_total: i64,
}

/// Sync runs kept per user.
pub const SYNC_RUNS_KEPT: i64 = 20;

/// Per-item results kept of a run, those with changes first. A run changes at
/// most this many items by default, but PDFs, merges, conflicts and failures
/// add results of their own, so its log can be cut short (`results_total`).
pub const MAX_SYNC_RESULTS: usize = 1000;

const SYNC_RUN_COLUMNS: &str = "user_id, started_at, finished_at, status, dry_run, \
     commonmeta_version, summary, error, results, results_total";

fn sync_run_row(r: &rusqlite::Row<'_>) -> rusqlite::Result<SyncRun> {
    let status: String = r.get(3)?;
    Ok(SyncRun {
        user_id: r.get::<_, i64>(0)? as u64,
        started_at: r.get(1)?,
        finished_at: r.get(2)?,
        status: RunStatus::parse(&status).unwrap_or(RunStatus::Failed),
        dry_run: r.get(4)?,
        commonmeta_version: r.get(5)?,
        summary: r.get(6)?,
        error: r.get(7)?,
        results: r.get(8)?,
        results_total: r.get(9)?,
    })
}

pub fn sync_setting(conn: &Connection, user_id: u64) -> rusqlite::Result<SyncSetting> {
    Ok(conn
        .query_row(
            "SELECT enabled, pdf, publications, deduplicate, overwrite FROM sync_setting
             WHERE user_id = ?1",
            [user_id as i64],
            |r| {
                Ok(SyncSetting {
                    enabled: r.get(0)?,
                    pdf: r.get(1)?,
                    publications: r.get(2)?,
                    deduplicate: r.get(3)?,
                    overwrite: r.get(4)?,
                })
            },
        )
        .optional()?
        .unwrap_or_default())
}

pub fn user_count(conn: &Connection) -> rusqlite::Result<i64> {
    conn.query_row("SELECT count(*) FROM user", [], |r| r.get(0))
}

/// Every library there is, with where its files are: the users' own, and the
/// groups', deleted ones included, whose data is kept.
pub fn libraries(conn: &Connection) -> rusqlite::Result<Vec<(LibraryRef, FileScope)>> {
    let mut stmt = conn.prepare("SELECT id, files_layout FROM user ORDER BY id")?;
    let rows = stmt.query_map([], |r| {
        let id = r.get::<_, i64>(0)? as u64;
        let legacy = r.get::<_, String>(1)? == "legacy";
        Ok((
            LibraryRef::User(id),
            if legacy {
                FileScope::Legacy
            } else {
                FileScope::User(id)
            },
        ))
    })?;
    let mut libraries = rows.collect::<rusqlite::Result<Vec<_>>>()?;
    let mut stmt = conn.prepare("SELECT id FROM grp ORDER BY id")?;
    let groups = stmt.query_map([], |r| {
        let id = r.get::<_, i64>(0)? as u64;
        Ok((LibraryRef::Group(id), FileScope::Group(id)))
    })?;
    libraries.extend(groups.collect::<rusqlite::Result<Vec<_>>>()?);
    Ok(libraries)
}

/// A member's role in a group.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub enum Role {
    Member,
    Admin,
    Owner,
}

impl Role {
    pub fn as_str(&self) -> &'static str {
        match self {
            Role::Member => "member",
            Role::Admin => "admin",
            Role::Owner => "owner",
        }
    }

    pub fn parse(text: &str) -> Option<Self> {
        match text {
            "member" => Some(Role::Member),
            "admin" => Some(Role::Admin),
            "owner" => Some(Role::Owner),
            _ => None,
        }
    }
}

/// A group library's settings, as Zotero has them.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Group {
    pub id: u64,
    pub name: String,
    pub description: String,
    /// Who may edit the library: `members` or `admins` (with the owner).
    pub library_editing: String,
    /// Who may add and change files: `members`, `admins` or `none`.
    pub file_editing: String,
    /// Bumped by every change to the group or its members, as Zotero's
    /// `format=versions` reports it.
    pub version: i64,
    pub created_at: i64,
    pub deleted: bool,
}

/// A member of a group, as the group's JSON and its page list them.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Member {
    pub user: User,
    pub role: Role,
}

/// What a user may do in a group library, from their role and its settings.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct GroupAccess {
    pub role: Role,
    pub write: bool,
    pub files: bool,
}

impl Group {
    /// What `role` may do here: anyone may read; editing as `library_editing`
    /// says; files as `file_editing` says, for those who may edit.
    pub fn access(&self, role: Role) -> GroupAccess {
        let manager = role >= Role::Admin;
        let write = manager || self.library_editing == "members";
        let files = write
            && match self.file_editing.as_str() {
                "members" => true,
                "admins" => manager,
                _ => false,
            };
        GroupAccess { role, write, files }
    }
}

const GROUP_COLUMNS: &str = "id, name, description, library_editing, file_editing, version, created_at, deleted_at IS NOT NULL";

fn group_row(r: &rusqlite::Row<'_>) -> rusqlite::Result<Group> {
    Ok(Group {
        id: r.get::<_, i64>(0)? as u64,
        name: r.get(1)?,
        description: r.get(2)?,
        library_editing: r.get(3)?,
        file_editing: r.get(4)?,
        version: r.get(5)?,
        created_at: r.get(6)?,
        deleted: r.get(7)?,
    })
}

pub fn group(conn: &Connection, id: u64) -> rusqlite::Result<Option<Group>> {
    conn.query_row(
        &format!("SELECT {GROUP_COLUMNS} FROM grp WHERE id = ?1"),
        [id as i64],
        group_row,
    )
    .optional()
}

pub fn role(conn: &Connection, group: u64, user: u64) -> rusqlite::Result<Option<Role>> {
    Ok(conn
        .query_row(
            "SELECT role FROM member WHERE group_id = ?1 AND user_id = ?2",
            [group as i64, user as i64],
            |r| r.get::<_, String>(0),
        )
        .optional()?
        .and_then(|role| Role::parse(&role)))
}

pub fn members(conn: &Connection, group: u64) -> rusqlite::Result<Vec<Member>> {
    let columns = USER_COLUMNS
        .split(", ")
        .map(|col| format!("u.{col}"))
        .collect::<Vec<_>>()
        .join(", ");
    let mut stmt = conn.prepare(&format!(
        "SELECT {columns}, m.role FROM member m JOIN user u ON u.id = m.user_id
         WHERE m.group_id = ?1
         ORDER BY CASE m.role WHEN 'owner' THEN 0 WHEN 'admin' THEN 1 ELSE 2 END, u.id"
    ))?;
    let rows = stmt.query_map([group as i64], |r| {
        Ok(Member {
            user: user_row(r)?,
            role: Role::parse(&r.get::<_, String>(USER_COLUMN_COUNT)?).unwrap_or(Role::Member),
        })
    })?;
    rows.collect()
}

fn bump_group(conn: &Connection, group: u64) -> rusqlite::Result<()> {
    conn.execute(
        "UPDATE grp SET version = version + 1 WHERE id = ?1",
        [group as i64],
    )?;
    Ok(())
}

/// Why a change to a group was refused.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum GroupRefusal {
    NoGroup,
    NoUser,
    /// The owner cannot leave, be removed or lose their role: the group would
    /// have none.
    Owner,
}

/// The highest version the library has served, 0 if none is known.
pub fn version_high(conn: &Connection, library: LibraryRef) -> rusqlite::Result<i64> {
    Ok(conn
        .query_row(
            "SELECT version_high FROM library_state WHERE library = ?1",
            [library.to_string()],
            |r| r.get(0),
        )
        .optional()?
        .unwrap_or(0))
}

pub fn note_version(conn: &Connection, library: LibraryRef, version: i64) -> rusqlite::Result<()> {
    conn.execute(
        "INSERT INTO library_state (library, version_high) VALUES (?1, ?2)
         ON CONFLICT (library) DO UPDATE SET version_high = excluded.version_high
         WHERE excluded.version_high > library_state.version_high",
        params![library.to_string(), version],
    )?;
    Ok(())
}

pub fn note_usage(conn: &Connection, library: LibraryRef, bytes: i64) -> rusqlite::Result<()> {
    conn.execute(
        "INSERT INTO library_state (library, bytes_used) VALUES (?1, ?2)
         ON CONFLICT (library) DO UPDATE SET bytes_used = excluded.bytes_used
         WHERE excluded.bytes_used != library_state.bytes_used",
        params![library.to_string(), bytes],
    )?;
    Ok(())
}

/// New groups get a random ID up to this, rather than the next one, so that
/// IDs tell nothing about how many there are or who came when. Users have
/// their zotero.org user ID.
const GROUP_ID_SPACE: u64 = 100_000;

/// A random ID for a new group: neither a group's nor, when `dir` is known, a
/// library file's there (which a control database restored from a backup may
/// not know about).
fn free_group_id(conn: &Connection, dir: Option<&Path>) -> rusqlite::Result<i64> {
    for _ in 0..1000 {
        let mut bytes = [0u8; 8];
        getrandom::fill(&mut bytes).expect("OS random number generator");
        let id = u64::from_le_bytes(bytes) % GROUP_ID_SPACE + 1;
        let used: bool = conn.query_row(
            "SELECT count(*) > 0 FROM grp WHERE id = ?1",
            [id as i64],
            |r| r.get(0),
        )?;
        let file =
            dir.is_some_and(|dir| crate::registry::path(dir, LibraryRef::Group(id)).exists());
        if !used && !file {
            return Ok(id as i64);
        }
    }
    Err(rusqlite::Error::SqliteFailure(
        rusqlite::ffi::Error::new(rusqlite::ffi::SQLITE_FULL),
        Some("no free group ID left".to_string()),
    ))
}

pub struct Control {
    pub db: Arc<Db>,
    /// Where the libraries' databases are, which new IDs must not collide with
    /// (set by `registry::Libraries::new`).
    pub libraries: std::sync::OnceLock<std::path::PathBuf>,
}

impl Control {
    /// Open (creating if needed) the control database at `path`.
    pub fn open(path: &Path) -> Result<Self, AppError> {
        Ok(Control {
            db: Db::open_with(path, db::CONTROL, 4)?,
            libraries: Default::default(),
        })
    }

    /// The installation's stable `Zotero-Server-ID`.
    pub async fn server_id(&self) -> Result<String, AppError> {
        self.db.server_id().await
    }

    pub async fn user(&self, id: u64) -> Result<Option<User>, AppError> {
        self.db.read(move |c| user(c, id)).await
    }

    /// Make `owner` the first user of a new server; anything else is refused
    /// once there are users, so a changed `--user-id` cannot quietly start an
    /// empty library.
    pub async fn ensure_owner(&self, owner: User) -> Result<(), AppError> {
        let id = owner.id;
        let outcome = self
            .db
            .write(move |c| {
                let tx = db::begin(c)?;
                if user(&tx, id)?.is_some() {
                    return Ok(None);
                }
                if user_count(&tx)? > 0 {
                    let ids: Vec<i64> = tx
                        .prepare("SELECT id FROM user ORDER BY id")?
                        .query_map([], |r| r.get(0))?
                        .collect::<rusqlite::Result<_>>()?;
                    return Ok(Some(ids));
                }
                insert_user(&tx, &owner)?;
                tx.commit()?;
                Ok(None)
            })
            .await?;
        match outcome {
            None => Ok(()),
            Some(ids) => Err(AppError::Internal(format!(
                "user {id} (DEFTER_USER_ID) is not among this server's users ({ids:?})"
            ))),
        }
    }

    pub async fn set_profile_hidden(&self, id: u64, hidden: bool) -> Result<(), AppError> {
        self.db
            .write(move |c| {
                c.execute(
                    "UPDATE user SET profile_hidden = ?2 WHERE id = ?1",
                    params![id as i64, hidden],
                )?;
                Ok(())
            })
            .await
    }

    pub async fn meta(&self, key: &'static str) -> Result<Option<String>, AppError> {
        self.db.read(move |c| db::get_meta(c, key)).await
    }

    pub async fn set_meta(&self, values: Vec<(&'static str, String)>) -> Result<(), AppError> {
        self.db
            .write(move |c| {
                let tx = db::begin(c)?;
                for (key, value) in &values {
                    db::set_meta(&tx, key, value)?;
                }
                tx.commit()
            })
            .await
    }

    pub async fn user_by_orcid(&self, orcid: &str) -> Result<Option<User>, AppError> {
        let orcid = orcid.to_string();
        self.db.read(move |c| user_by_orcid(c, &orcid)).await
    }

    /// Bring what the settings say into the database: `owner_orcid`
    /// (`--orcid`) is the owner's.
    pub async fn reconcile(&self, owner: u64, owner_orcid: Option<String>) -> Result<(), AppError> {
        let Some(orcid) = owner_orcid else {
            return Ok(());
        };
        let outcome = self
            .db
            .write(move |c| {
                let tx = db::begin(c)?;
                match user_by_orcid(&tx, &orcid)? {
                    Some(user) if user.id != owner => return Ok(Some(user.id)),
                    Some(_) => {}
                    None => {
                        tx.execute(
                            "UPDATE user SET orcid = ?2 WHERE id = ?1",
                            params![owner as i64, orcid],
                        )?;
                    }
                }
                tx.commit()?;
                Ok(None)
            })
            .await?;
        match outcome {
            None => Ok(()),
            Some(other) => Err(AppError::Internal(format!(
                "--orcid belongs to user {other}, not to the owner"
            ))),
        }
    }

    /// Sign in the account with `orcid`: its user, else `NoAccount` if invited
    /// or an `admin` (on `--admin-orcids`), who then join.
    ///
    /// With a `grant` from the provider, it decides instead of invitations, at
    /// every sign-in: a user it no longer grants access is disabled (their keys
    /// revoked) unless they are an admin or the `owner`, and enabled again once
    /// it does, unless an admin disabled them. Their `provider_admin` is what
    /// it says now. A grant `missing` the claim refuses all but admins and the
    /// owner, and changes nobody.
    pub async fn sign_in(
        &self,
        orcid: &str,
        admin: bool,
        grant: Option<Grant>,
        owner: u64,
    ) -> Result<SignIn, AppError> {
        let orcid = orcid.to_string();
        self.db
            .write(move |c| {
                let tx = db::begin(c)?;
                let found = user_by_orcid(&tx, &orcid)?;
                let Some(grant) = grant else {
                    return Ok(match found {
                        Some(user) if user.disabled => SignIn::Disabled,
                        Some(user) => SignIn::User(user),
                        None if admin || invited(&tx, &orcid)? => SignIn::NoAccount,
                        None => SignIn::NotInvited,
                    });
                };
                let Some(mut user) = found else {
                    return Ok(if admin || grant.user {
                        SignIn::NoAccount
                    } else {
                        SignIn::NoAccess { disabled: None }
                    });
                };
                if grant.missing {
                    // Nothing is known: nothing changes, not even
                    // `provider_admin`.
                    return Ok(if !(admin || user.id == owner) {
                        SignIn::NoAccess { disabled: None }
                    } else if user.disabled {
                        SignIn::Disabled
                    } else {
                        SignIn::User(user)
                    });
                }
                if !(admin || grant.user || user.id == owner) {
                    if user.disabled {
                        // By an admin, or by an earlier sign-in: it stays so.
                        return Ok(SignIn::NoAccess { disabled: None });
                    }
                    disable(&tx, user.id, true)?;
                    tx.commit()?;
                    return Ok(SignIn::NoAccess {
                        disabled: Some(user.id),
                    });
                }
                if user.disabled && !user.access_disabled {
                    return Ok(SignIn::Disabled);
                }
                // Written only when it changes: every write is a revision to
                // back up.
                if user.disabled || user.provider_admin != grant.admin {
                    tx.execute(
                        "UPDATE user SET disabled = 0, access_disabled = 0, provider_admin = ?2
                         WHERE id = ?1",
                        params![user.id as i64, grant.admin],
                    )?;
                    tx.commit()?;
                    user.disabled = false;
                    user.access_disabled = false;
                    user.provider_admin = grant.admin;
                }
                Ok(SignIn::User(user))
            })
            .await
    }

    /// Make the account with `orcid`, invited or an admin, a user with `id`, its
    /// zotero.org user ID, and the zotero.org `username` (else its iD, when
    /// another user here has that name), using up its invitation. When `id` is
    /// the `owner`'s, who has no iD yet, an admin's account becomes the owner,
    /// with the zotero.org `username` unless another user here has it.
    /// A user it became meanwhile signs in as such; `IdTaken` when `id` is
    /// another user's, or a library file nobody here owns has it. With a
    /// `grant` from the provider, it decides instead of the invitation, as in
    /// [`Control::sign_in`].
    pub async fn join(
        &self,
        orcid: &str,
        admin: bool,
        grant: Option<Grant>,
        id: u64,
        username: &str,
        owner: u64,
    ) -> Result<SignIn, AppError> {
        let (orcid, username) = (orcid.to_string(), username.to_string());
        let dir = self.libraries.get().cloned();
        self.db
            .write(move |c| {
                let tx = db::begin(c)?;
                if let Some(user) = user_by_orcid(&tx, &orcid)? {
                    return Ok(if user.disabled {
                        SignIn::Disabled
                    } else {
                        SignIn::User(user)
                    });
                }
                let provider_admin = grant.is_some_and(|g| g.admin);
                match grant {
                    None if !admin && !invited(&tx, &orcid)? => return Ok(SignIn::NotInvited),
                    Some(grant) if !admin && !grant.user => {
                        return Ok(SignIn::NoAccess { disabled: None });
                    }
                    _ => {}
                }
                // The owner of a server from before signs in for the first
                // time, as an admin: their zotero.org user ID is DEFTER_USER_ID.
                // Anyone else holding the owner's key gets `IdTaken`.
                if (admin || provider_admin)
                    && id == owner
                    && let Some(mut owner) = user(&tx, owner)?
                    && owner.orcid.is_none()
                {
                    if !username.is_empty() && !name_taken(&tx, &username, owner.id)? {
                        owner.username = username;
                    }
                    tx.execute(
                        "UPDATE user SET orcid = ?2, provider_admin = ?3, username = ?4
                         WHERE id = ?1",
                        params![owner.id as i64, orcid, provider_admin, owner.username],
                    )?;
                    tx.execute(
                        "UPDATE invite SET used_by = ?2, used_at = ?3 WHERE orcid = ?1",
                        params![orcid, owner.id as i64, now()],
                    )?;
                    tx.commit()?;
                    owner.orcid = Some(orcid);
                    owner.provider_admin = provider_admin;
                    return Ok(if owner.disabled {
                        SignIn::Disabled
                    } else {
                        SignIn::User(owner)
                    });
                }
                let file = dir
                    .is_some_and(|dir| crate::registry::path(&dir, LibraryRef::User(id)).exists());
                if id == 0 || id > i64::MAX as u64 || user(&tx, id)?.is_some() || file {
                    return Ok(SignIn::IdTaken);
                }
                let taken = name_taken(&tx, &username, id)?;
                let user = User {
                    id,
                    orcid: Some(orcid.clone()),
                    username: if username.is_empty() || taken {
                        orcid.clone()
                    } else {
                        username
                    },
                    display_name: None,
                    // Shown once they choose to.
                    profile_hidden: true,
                    disabled: false,
                    legacy_files: false,
                    quota_bytes: None,
                    access_disabled: false,
                    provider_admin,
                };
                insert_user(&tx, &user)?;
                tx.execute(
                    "UPDATE invite SET used_by = ?2, used_at = ?3 WHERE orcid = ?1",
                    params![orcid, id as i64, now()],
                )?;
                tx.commit()?;
                Ok(SignIn::User(user))
            })
            .await
    }

    /// Give user `id` the user name `username`, as `--username` names the
    /// owner; refused when another user here has it.
    pub async fn set_username(&self, id: u64, username: String) -> Result<(), AppError> {
        let name = username.clone();
        let set = self
            .db
            .write(move |c| {
                let tx = db::begin(c)?;
                if name_taken(&tx, &name, id)? {
                    return Ok(false);
                }
                tx.execute(
                    "UPDATE user SET username = ?2 WHERE id = ?1",
                    params![id as i64, name],
                )?;
                tx.commit()?;
                Ok(true)
            })
            .await?;
        if set {
            Ok(())
        } else {
            Err(AppError::BadRequest(format!(
                "another user has the user name {username:?}"
            )))
        }
    }

    pub async fn set_display_name(&self, id: u64, name: String) -> Result<(), AppError> {
        self.db
            .write(move |c| {
                c.execute(
                    "UPDATE user SET display_name = ?2 WHERE id = ?1",
                    params![id as i64, name],
                )?;
                Ok(())
            })
            .await
    }

    pub async fn set_quota(&self, id: u64, quota_bytes: Option<i64>) -> Result<(), AppError> {
        self.db
            .write(move |c| {
                c.execute(
                    "UPDATE user SET quota_bytes = ?2 WHERE id = ?1",
                    params![id as i64, quota_bytes],
                )?;
                Ok(())
            })
            .await
    }

    /// Disable or enable a user, as an admin: disabled so, signing in does
    /// not enable them again. Disabling also revokes their keys, so their
    /// devices stop syncing at once and need a new login once enabled again.
    pub async fn set_disabled(&self, id: u64, disabled: bool) -> Result<(), AppError> {
        self.db
            .write(move |c| {
                let tx = db::begin(c)?;
                if disabled {
                    self::disable(&tx, id, false)?;
                } else {
                    tx.execute(
                        "UPDATE user SET disabled = 0, access_disabled = 0 WHERE id = ?1",
                        [id as i64],
                    )?;
                }
                tx.commit()
            })
            .await
    }

    /// Forget who the provider said was an admin, when no `--oidc-admin-values`
    /// are set: should they be set again, only a sign-in then makes someone an
    /// admin, not one from before. The number of users it was.
    pub async fn forget_provider_admins(&self) -> Result<usize, AppError> {
        self.db
            .write(|c| {
                c.execute(
                    "UPDATE user SET provider_admin = 0 WHERE provider_admin = 1",
                    [],
                )
            })
            .await
    }

    /// Every user, with their usage and devices, for admins.
    pub async fn users(&self) -> Result<Vec<UserOverview>, AppError> {
        self.db
            .read(|c| {
                let columns = USER_COLUMNS
                    .split(", ")
                    .map(|col| format!("u.{col}"))
                    .collect::<Vec<_>>()
                    .join(", ");
                let mut stmt = c.prepare(&format!(
                    "SELECT {columns},
                            coalesce((SELECT sum(bytes_used) FROM library_state
                                      WHERE library = 'u' || u.id OR library IN (
                                          SELECT 'g' || m.group_id FROM member m
                                          JOIN grp g ON g.id = m.group_id
                                          WHERE m.user_id = u.id AND m.role = 'owner'
                                            AND g.deleted_at IS NULL)), 0),
                            (SELECT count(*) FROM api_key k
                             WHERE k.user_id = u.id AND k.revoked_at IS NULL)
                     FROM user u ORDER BY u.id"
                ))?;
                let rows = stmt.query_map([], |r| {
                    Ok(UserOverview {
                        user: user_row(r)?,
                        bytes_used: r.get(USER_COLUMN_COUNT)?,
                        keys: r.get(USER_COLUMN_COUNT + 1)?,
                    })
                })?;
                rows.collect()
            })
            .await
    }

    pub async fn invites(&self) -> Result<Vec<Invite>, AppError> {
        self.db
            .read(|c| {
                let mut stmt = c.prepare(
                    "SELECT orcid, note, created_at, used_by FROM invite
                     ORDER BY used_by IS NOT NULL, created_at DESC, orcid",
                )?;
                let rows = stmt.query_map([], |r| {
                    Ok(Invite {
                        orcid: r.get(0)?,
                        note: r.get(1)?,
                        created_at: r.get(2)?,
                        used_by: r.get::<_, Option<i64>>(3)?.map(|id| id as u64),
                    })
                })?;
                rows.collect()
            })
            .await
    }

    /// Invite `orcid`. False if they are a user already.
    pub async fn invite(
        &self,
        orcid: String,
        by: u64,
        note: Option<String>,
    ) -> Result<bool, AppError> {
        self.db
            .write(move |c| {
                let tx = db::begin(c)?;
                if user_by_orcid(&tx, &orcid)?.is_some() {
                    return Ok(false);
                }
                tx.execute(
                    "INSERT INTO invite (orcid, invited_by, note, created_at) VALUES (?1, ?2, ?3, ?4)
                     ON CONFLICT (orcid) DO UPDATE SET note = excluded.note
                     WHERE invite.used_by IS NULL",
                    params![orcid, by as i64, note, now()],
                )?;
                tx.commit()?;
                Ok(true)
            })
            .await
    }

    /// Take back an invitation not used yet.
    pub async fn uninvite(&self, orcid: String) -> Result<(), AppError> {
        self.db
            .write(move |c| {
                c.execute(
                    "DELETE FROM invite WHERE orcid = ?1 AND used_by IS NULL",
                    [orcid],
                )?;
                Ok(())
            })
            .await
    }

    /// A new key for `user_id`'s device, named `name`. Only its hash is kept:
    /// the key itself is returned once.
    pub async fn mint_key(
        &self,
        user_id: u64,
        name: String,
        write: bool,
    ) -> Result<String, AppError> {
        let key = crate::auth::random_hex(32);
        let (hash, prefix) = (key_hash(&key), key[..8].to_string());
        self.db
            .write(move |c| {
                c.execute(
                    "INSERT INTO api_key (user_id, hash, prefix, name, write, created_at)
                     VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
                    params![user_id as i64, hash, prefix, name, write, now()],
                )?;
                Ok(())
            })
            .await?;
        Ok(key)
    }

    /// Whose minted key `key` is, if it is live and its user enabled. Records
    /// its use, at most once per [`KEY_USE_PRECISION`].
    pub async fn key_owner(&self, key: &str) -> Result<Option<KeyOwner>, AppError> {
        let hash = key_hash(key);
        let found = self
            .db
            .read(move |c| {
                c.query_row(
                    "SELECT k.id, k.user_id, k.write, coalesce(k.last_used_at, 0)
                     FROM api_key k JOIN user u ON u.id = k.user_id
                     WHERE k.hash = ?1 AND k.revoked_at IS NULL AND u.disabled = 0",
                    [hash],
                    |r| {
                        Ok((
                            KeyOwner {
                                key_id: r.get(0)?,
                                user_id: r.get::<_, i64>(1)? as u64,
                                write: r.get(2)?,
                            },
                            r.get::<_, i64>(3)?,
                        ))
                    },
                )
                .optional()
            })
            .await?;
        let Some((owner, last_used)) = found else {
            return Ok(None);
        };
        let at = now();
        if at - last_used >= KEY_USE_PRECISION {
            let id = owner.key_id;
            self.db
                .write(move |c| {
                    c.execute(
                        "UPDATE api_key SET last_used_at = ?2 WHERE id = ?1",
                        params![id, at],
                    )
                })
                .await?;
        }
        Ok(Some(owner))
    }

    /// `user_id`'s keys that are not revoked, newest first.
    pub async fn keys(&self, user_id: u64) -> Result<Vec<ApiKey>, AppError> {
        self.db
            .read(move |c| {
                let mut stmt = c.prepare(
                    "SELECT id, prefix, name, write, created_at, last_used_at FROM api_key
                     WHERE user_id = ?1 AND revoked_at IS NULL ORDER BY created_at DESC, id DESC",
                )?;
                let rows = stmt.query_map([user_id as i64], |r| {
                    Ok(ApiKey {
                        id: r.get(0)?,
                        prefix: r.get(1)?,
                        name: r.get(2)?,
                        write: r.get(3)?,
                        created_at: r.get(4)?,
                        last_used_at: r.get(5)?,
                    })
                })?;
                rows.collect()
            })
            .await
    }

    /// Revoke `user_id`'s key `key_id`; false if they have no such live key.
    pub async fn revoke_key(&self, user_id: u64, key_id: i64) -> Result<bool, AppError> {
        self.db
            .write(move |c| {
                Ok(c.execute(
                    "UPDATE api_key SET revoked_at = ?3
                     WHERE id = ?1 AND user_id = ?2 AND revoked_at IS NULL",
                    params![key_id, user_id as i64, now()],
                )? > 0)
            })
            .await
    }

    pub async fn note_usage(&self, library: LibraryRef, bytes: i64) -> Result<(), AppError> {
        self.db.write(move |c| note_usage(c, library, bytes)).await
    }

    /// A new group owned by `owner`, named `name`, editable by its members.
    pub async fn create_group(&self, owner: u64, name: String) -> Result<Group, AppError> {
        let dir = self.libraries.get().cloned();
        self.db
            .write(move |c| {
                let tx = db::begin(c)?;
                let id = free_group_id(&tx, dir.as_deref())?;
                tx.execute(
                    "INSERT INTO grp (id, name, version, created_at) VALUES (?1, ?2, 1, ?3)",
                    params![id, name, now()],
                )?;
                tx.execute(
                    "INSERT INTO member (group_id, user_id, role) VALUES (?1, ?2, 'owner')",
                    params![id, owner as i64],
                )?;
                let created = group(&tx, id as u64)?;
                tx.commit()?;
                created.ok_or(rusqlite::Error::QueryReturnedNoRows)
            })
            .await
    }

    /// A group, deleted or not.
    pub async fn group(&self, id: u64) -> Result<Option<Group>, AppError> {
        self.db.read(move |c| group(c, id)).await
    }

    /// A group that is not deleted, with its members.
    pub async fn group_with_members(
        &self,
        id: u64,
    ) -> Result<Option<(Group, Vec<Member>)>, AppError> {
        self.db
            .read(move |c| match group(c, id)? {
                Some(g) if !g.deleted => Ok(Some((g, members(c, id)?))),
                _ => Ok(None),
            })
            .await
    }

    /// The groups `user` is in, not deleted, with their role.
    pub async fn groups_of(&self, user: u64) -> Result<Vec<(Group, Role)>, AppError> {
        self.db
            .read(move |c| {
                let columns = GROUP_COLUMNS
                    .split(", ")
                    .map(|col| format!("g.{col}"))
                    .collect::<Vec<_>>()
                    .join(", ");
                let mut stmt = c.prepare(&format!(
                    "SELECT {columns}, m.role FROM grp g JOIN member m ON m.group_id = g.id
                     WHERE m.user_id = ?1 AND g.deleted_at IS NULL ORDER BY g.name, g.id"
                ))?;
                let rows = stmt.query_map([user as i64], |r| {
                    Ok((
                        group_row(r)?,
                        Role::parse(&r.get::<_, String>(8)?).unwrap_or(Role::Member),
                    ))
                })?;
                rows.collect()
            })
            .await
    }

    /// What `user` may do in group `id`; `None` when they are not a member or it
    /// is deleted.
    pub async fn group_access(&self, id: u64, user: u64) -> Result<Option<GroupAccess>, AppError> {
        self.db
            .read(move |c| {
                let Some(group) = group(c, id)?.filter(|g| !g.deleted) else {
                    return Ok(None);
                };
                Ok(role(c, id, user)?.map(|role| group.access(role)))
            })
            .await
    }

    /// Change a group's name, description or who may edit it, each when given.
    pub async fn update_group(
        &self,
        id: u64,
        name: Option<String>,
        description: Option<String>,
        library_editing: Option<String>,
        file_editing: Option<String>,
    ) -> Result<(), AppError> {
        self.db
            .write(move |c| {
                let tx = db::begin(c)?;
                tx.execute(
                    "UPDATE grp SET name = coalesce(?2, name),
                                    description = coalesce(?3, description),
                                    library_editing = coalesce(?4, library_editing),
                                    file_editing = coalesce(?5, file_editing)
                     WHERE id = ?1 AND deleted_at IS NULL",
                    params![id as i64, name, description, library_editing, file_editing],
                )?;
                bump_group(&tx, id)?;
                tx.commit()
            })
            .await
    }

    /// Make `user` a member of group `id` with `role` (admin or member), or
    /// change their role. Not for the owner.
    pub async fn set_member(
        &self,
        id: u64,
        user: u64,
        role: Role,
    ) -> Result<Result<(), GroupRefusal>, AppError> {
        self.db
            .write(move |c| {
                let tx = db::begin(c)?;
                if group(&tx, id)?.is_none_or(|g| g.deleted) {
                    return Ok(Err(GroupRefusal::NoGroup));
                }
                if self::user(&tx, user)?.is_none_or(|u| u.disabled) {
                    return Ok(Err(GroupRefusal::NoUser));
                }
                if role == Role::Owner || self::role(&tx, id, user)? == Some(Role::Owner) {
                    return Ok(Err(GroupRefusal::Owner));
                }
                tx.execute(
                    "INSERT INTO member (group_id, user_id, role) VALUES (?1, ?2, ?3)
                     ON CONFLICT (group_id, user_id) DO UPDATE SET role = excluded.role",
                    params![id as i64, user as i64, role.as_str()],
                )?;
                bump_group(&tx, id)?;
                tx.commit()?;
                Ok(Ok(()))
            })
            .await
    }

    /// Take `user` out of group `id`. Not the owner.
    pub async fn remove_member(
        &self,
        id: u64,
        user: u64,
    ) -> Result<Result<(), GroupRefusal>, AppError> {
        self.db
            .write(move |c| {
                let tx = db::begin(c)?;
                match role(&tx, id, user)? {
                    None => return Ok(Err(GroupRefusal::NoUser)),
                    Some(Role::Owner) => return Ok(Err(GroupRefusal::Owner)),
                    Some(_) => {}
                }
                tx.execute(
                    "DELETE FROM member WHERE group_id = ?1 AND user_id = ?2",
                    params![id as i64, user as i64],
                )?;
                bump_group(&tx, id)?;
                tx.commit()?;
                Ok(Ok(()))
            })
            .await
    }

    /// Delete group `id`: its members' clients are told it is gone. Its library
    /// is kept, and backed up, but nobody can reach it.
    pub async fn delete_group(&self, id: u64) -> Result<(), AppError> {
        self.db
            .write(move |c| {
                let tx = db::begin(c)?;
                tx.execute(
                    "UPDATE grp SET deleted_at = ?2 WHERE id = ?1 AND deleted_at IS NULL",
                    params![id as i64, now()],
                )?;
                bump_group(&tx, id)?;
                tx.commit()
            })
            .await
    }

    /// The bytes `user`'s attachment files take as last counted, in their own
    /// library and the groups they own, but `except`, which is counted live. A
    /// deleted group's files count no more.
    pub async fn usage_of(&self, user: u64, except: LibraryRef) -> Result<i64, AppError> {
        self.db
            .read(move |c| {
                c.query_row(
                    "SELECT coalesce(sum(bytes_used), 0) FROM library_state
                     WHERE library != ?2 AND (library = 'u' || ?1 OR library IN (
                         SELECT 'g' || m.group_id FROM member m JOIN grp g ON g.id = m.group_id
                         WHERE m.user_id = ?1 AND m.role = 'owner' AND g.deleted_at IS NULL))",
                    params![user as i64, except.to_string()],
                    |r| r.get(0),
                )
            })
            .await
    }

    /// The owner of group `id`.
    pub async fn group_owner(&self, id: u64) -> Result<Option<User>, AppError> {
        self.db
            .read(move |c| {
                let owner: Option<i64> = c
                    .query_row(
                        "SELECT user_id FROM member WHERE group_id = ?1 AND role = 'owner'",
                        [id as i64],
                        |r| r.get(0),
                    )
                    .optional()?;
                match owner {
                    Some(owner) => user(c, owner as u64),
                    None => Ok(None),
                }
            })
            .await
    }

    pub async fn version_high(&self, library: LibraryRef) -> Result<i64, AppError> {
        self.db.read(move |c| version_high(c, library)).await
    }

    /// What `user_id` chose for the server-side sync; all off when they never
    /// chose.
    pub async fn sync_setting(&self, user_id: u64) -> Result<SyncSetting, AppError> {
        self.db.read(move |c| sync_setting(c, user_id)).await
    }

    pub async fn set_sync_setting(
        &self,
        user_id: u64,
        setting: SyncSetting,
    ) -> Result<(), AppError> {
        self.db
            .write(move |c| {
                let tx = db::begin(c)?;
                tx.execute(
                    "INSERT INTO sync_setting
                         (user_id, enabled, pdf, publications, deduplicate, overwrite,
                          changed_at)
                     VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)
                     ON CONFLICT (user_id) DO UPDATE SET
                         enabled = excluded.enabled, pdf = excluded.pdf,
                         publications = excluded.publications,
                         deduplicate = excluded.deduplicate, overwrite = excluded.overwrite,
                         changed_at = excluded.changed_at",
                    params![
                        user_id as i64,
                        setting.enabled,
                        setting.pdf,
                        setting.publications,
                        setting.deduplicate,
                        setting.overwrite,
                        now()
                    ],
                )?;
                tx.commit()
            })
            .await
    }

    /// `user_id`'s sync setting while they opted in and are enabled; `None`
    /// otherwise. Asked again before every write of a run, so opting out or
    /// being disabled stops it at once.
    pub async fn sync_access(&self, user_id: u64) -> Result<Option<SyncSetting>, AppError> {
        self.db
            .read(move |c| {
                let Some(user) = user(c, user_id)? else {
                    return Ok(None);
                };
                let setting = sync_setting(c, user_id)?;
                Ok((setting.enabled && !user.disabled).then_some(setting))
            })
            .await
    }

    /// The users to sync, with their settings: those who opted in and are
    /// enabled, `owner` first, then whoever waited longest since their last run.
    pub async fn sync_jobs(&self, owner: u64) -> Result<Vec<(User, SyncSetting)>, AppError> {
        self.db
            .read(move |c| {
                let columns = USER_COLUMNS
                    .split(", ")
                    .map(|col| format!("u.{col}"))
                    .collect::<Vec<_>>()
                    .join(", ");
                let mut stmt = c.prepare(&format!(
                    "SELECT {columns}, s.pdf, s.publications, s.deduplicate, s.overwrite
                     FROM user u JOIN sync_setting s ON s.user_id = u.id
                     WHERE s.enabled AND NOT u.disabled
                     ORDER BY u.id != ?1,
                              coalesce((SELECT max(started_at) FROM sync_run r
                                        WHERE r.user_id = u.id), 0),
                              u.id"
                ))?;
                let rows = stmt.query_map([owner as i64], |r| {
                    Ok((
                        user_row(r)?,
                        SyncSetting {
                            enabled: true,
                            pdf: r.get(USER_COLUMN_COUNT)?,
                            publications: r.get(USER_COLUMN_COUNT + 1)?,
                            deduplicate: r.get(USER_COLUMN_COUNT + 2)?,
                            overwrite: r.get(USER_COLUMN_COUNT + 3)?,
                        },
                    ))
                })?;
                rows.collect()
            })
            .await
    }

    /// Keep a finished run. Older runs of the user lose their results, and
    /// only the newest [`SYNC_RUNS_KEPT`] are kept.
    pub async fn record_run(&self, run: SyncRun) -> Result<(), AppError> {
        self.db
            .write(move |c| {
                let tx = db::begin(c)?;
                let user_id = run.user_id as i64;
                tx.execute(
                    &format!(
                        "INSERT INTO sync_run ({SYNC_RUN_COLUMNS})
                         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)"
                    ),
                    params![
                        user_id,
                        run.started_at,
                        run.finished_at,
                        run.status.as_str(),
                        run.dry_run,
                        run.commonmeta_version,
                        run.summary,
                        run.error,
                        run.results,
                        run.results_total
                    ],
                )?;
                let id = tx.last_insert_rowid();
                tx.execute(
                    "UPDATE sync_run SET results = NULL
                     WHERE user_id = ?1 AND id != ?2 AND results IS NOT NULL",
                    params![user_id, id],
                )?;
                tx.execute(
                    "DELETE FROM sync_run WHERE user_id = ?1 AND id NOT IN (
                         SELECT id FROM sync_run WHERE user_id = ?1
                         ORDER BY id DESC LIMIT ?2)",
                    params![user_id, SYNC_RUNS_KEPT],
                )?;
                tx.commit()
            })
            .await
    }

    /// `user_id`'s newest run, with its results.
    pub async fn last_run(&self, user_id: u64) -> Result<Option<SyncRun>, AppError> {
        self.db
            .read(move |c| {
                c.query_row(
                    &format!(
                        "SELECT {SYNC_RUN_COLUMNS} FROM sync_run WHERE user_id = ?1
                         ORDER BY id DESC LIMIT 1"
                    ),
                    [user_id as i64],
                    sync_run_row,
                )
                .optional()
            })
            .await
    }

    /// Every user's newest run, without results, by user, for admins.
    pub async fn last_runs(&self) -> Result<HashMap<u64, SyncRun>, AppError> {
        self.db
            .read(|c| {
                let columns = SYNC_RUN_COLUMNS.replace("results,", "NULL,");
                let mut stmt = c.prepare(&format!(
                    "SELECT {columns} FROM sync_run
                     WHERE id IN (SELECT max(id) FROM sync_run GROUP BY user_id)"
                ))?;
                let rows = stmt.query_map([], sync_run_row)?;
                rows.map(|run| run.map(|run| (run.user_id, run))).collect()
            })
            .await
    }

    /// Every user's sync setting, by user, for admins.
    pub async fn sync_settings(&self) -> Result<HashMap<u64, SyncSetting>, AppError> {
        self.db
            .read(|c| {
                let mut stmt = c.prepare(
                    "SELECT user_id, enabled, pdf, publications, deduplicate, overwrite
                     FROM sync_setting",
                )?;
                let rows = stmt.query_map([], |r| {
                    Ok((
                        r.get::<_, i64>(0)? as u64,
                        SyncSetting {
                            enabled: r.get(1)?,
                            pdf: r.get(2)?,
                            publications: r.get(3)?,
                            deduplicate: r.get(4)?,
                            overwrite: r.get(5)?,
                        },
                    ))
                })?;
                rows.collect()
            })
            .await
    }

    pub async fn note_version(&self, library: LibraryRef, version: i64) -> Result<(), AppError> {
        self.db
            .write(move |c| note_version(c, library, version))
            .await
    }
}

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

    fn owner(id: u64) -> User {
        User {
            id,
            orcid: None,
            username: "owner".into(),
            display_name: None,
            profile_hidden: false,
            disabled: false,
            legacy_files: false,
            quota_bytes: None,
            access_disabled: false,
            provider_admin: false,
        }
    }

    #[test]
    fn names_libraries() {
        assert_eq!(LibraryRef::User(12).to_string(), "u12");
        assert_eq!(LibraryRef::parse("g3"), Some(LibraryRef::Group(3)));
        assert_eq!(LibraryRef::parse("u12"), Some(LibraryRef::User(12)));
        for bad in ["", "u", "x1", "u-1", "u+1", "u1.sqlite3", "control"] {
            assert_eq!(LibraryRef::parse(bad), None, "{bad}");
        }
    }

    #[tokio::test]
    async fn keeps_users_and_what_libraries_served() {
        let dir = tempfile::tempdir().unwrap();
        let control = Control::open(&dir.path().join("control.sqlite3")).unwrap();
        control.ensure_owner(owner(7)).await.unwrap();
        // Again is fine; another owner once there are users is not.
        control.ensure_owner(owner(7)).await.unwrap();
        let refused = control.ensure_owner(owner(8)).await.unwrap_err();
        assert!(refused.to_string().contains("[7]"), "{refused}");

        let user = control.user(7).await.unwrap().unwrap();
        assert_eq!(user.files(), FileScope::User(7));
        control.set_profile_hidden(7, true).await.unwrap();
        control
            .reconcile(7, Some("0000-0003-1419-2405".into()))
            .await
            .unwrap();
        assert_eq!(control.user(7).await.unwrap().unwrap().name(), "owner");
        control
            .set_display_name(7, "Ada Example".into())
            .await
            .unwrap();
        let user = control.user(7).await.unwrap().unwrap();
        assert_eq!(user.name(), "Ada Example");
        assert!(user.profile_hidden);
        assert_eq!(user.orcid.as_deref(), Some("0000-0003-1419-2405"));

        let library = LibraryRef::User(7);
        assert_eq!(control.version_high(library).await.unwrap(), 0);
        control.note_version(library, 5).await.unwrap();
        // It never goes down.
        control.note_version(library, 3).await.unwrap();
        assert_eq!(control.version_high(library).await.unwrap(), 5);
    }

    #[tokio::test]
    async fn the_providers_grant_disables_and_enables_at_sign_in() {
        const ALICE: &str = "0000-0001-5109-3700";
        let dir = tempfile::tempdir().unwrap();
        let control = Control::open(&dir.path().join("control.sqlite3")).unwrap();
        control.ensure_owner(owner(1)).await.unwrap();
        let (none, user, admin) = (
            Some(Grant::default()),
            Some(Grant {
                user: true,
                admin: false,
                ..Grant::default()
            }),
            Some(Grant {
                user: true,
                admin: true,
                ..Grant::default()
            }),
        );
        let sign_in = |grant| control.sign_in(ALICE, false, grant, 1);
        assert_eq!(
            sign_in(none).await.unwrap(),
            SignIn::NoAccess { disabled: None }
        );
        assert_eq!(sign_in(user).await.unwrap(), SignIn::NoAccount);
        // Joining checks the grant again, not the invitation.
        assert_eq!(
            control
                .join(ALICE, false, none, 5, "alice", 1)
                .await
                .unwrap(),
            SignIn::NoAccess { disabled: None }
        );
        let SignIn::User(alice) = control
            .join(ALICE, false, admin, 5, "alice", 1)
            .await
            .unwrap()
        else {
            panic!("she may join");
        };
        assert!(alice.provider_admin);
        let SignIn::User(alice) = sign_in(user).await.unwrap() else {
            panic!("a user");
        };
        assert!(!alice.provider_admin);
        control.mint_key(5, "laptop".into(), true).await.unwrap();

        let revision = || async {
            control
                .db
                .read(|c| c.query_row("SELECT n FROM revision", [], |r| r.get::<_, i64>(0)))
                .await
                .unwrap()
        };
        // Nothing changes, nothing is written.
        let before = revision().await;
        sign_in(user).await.unwrap();
        assert_eq!(revision().await, before);

        assert_eq!(
            sign_in(none).await.unwrap(),
            SignIn::NoAccess { disabled: Some(5) }
        );
        assert!(control.keys(5).await.unwrap().is_empty());
        assert_eq!(
            sign_in(none).await.unwrap(),
            SignIn::NoAccess { disabled: None },
            "disabled already"
        );
        assert!(matches!(sign_in(admin).await.unwrap(), SignIn::User(u) if u.provider_admin));
        // Forgotten when admin values are no longer set; nothing written
        // when there is nothing to forget.
        assert_eq!(control.forget_provider_admins().await.unwrap(), 1);
        assert!(!control.user(5).await.unwrap().unwrap().provider_admin);
        let before = revision().await;
        assert_eq!(control.forget_provider_admins().await.unwrap(), 0);
        assert_eq!(revision().await, before);
        assert!(matches!(sign_in(admin).await.unwrap(), SignIn::User(u) if u.provider_admin));
        // Without the claim, she is refused, and nothing changes.
        let missing = Some(Grant {
            missing: true,
            ..Grant::default()
        });
        control.mint_key(5, "laptop".into(), true).await.unwrap();
        let before = revision().await;
        assert_eq!(
            sign_in(missing).await.unwrap(),
            SignIn::NoAccess { disabled: None }
        );
        assert_eq!(revision().await, before);
        assert_eq!(control.keys(5).await.unwrap().len(), 1);
        assert!(matches!(
            control.sign_in(ALICE, true, missing, 1).await.unwrap(),
            SignIn::User(u) if u.provider_admin
        ));
        assert_eq!(
            control
                .sign_in("0000-0002-1825-0097", false, missing, 1)
                .await
                .unwrap(),
            SignIn::NoAccess { disabled: None }
        );
        // An admin's disabling sticks; enabling hands back to the provider.
        control.set_disabled(5, true).await.unwrap();
        assert_eq!(sign_in(user).await.unwrap(), SignIn::Disabled);
        control.set_disabled(5, false).await.unwrap();
        assert!(matches!(sign_in(user).await.unwrap(), SignIn::User(_)));
        // The owner and admins on --admin-orcids are never disabled so.
        control
            .reconcile(1, Some("0000-0003-1419-2405".into()))
            .await
            .unwrap();
        assert!(matches!(
            control.sign_in("0000-0003-1419-2405", false, none, 1).await.unwrap(),
            SignIn::User(u) if u.id == 1
        ));
        assert!(matches!(
            control.sign_in(ALICE, true, none, 1).await.unwrap(),
            SignIn::User(u) if !u.provider_admin
        ));
        // Without a grant, invitations decide, as before.
        assert_eq!(
            control
                .sign_in("0000-0002-1825-0097", false, None, 1)
                .await
                .unwrap(),
            SignIn::NotInvited
        );
    }

    #[tokio::test]
    async fn counts_every_change() {
        let dir = tempfile::tempdir().unwrap();
        let control = Control::open(&dir.path().join("control.sqlite3")).unwrap();
        let revision = || async {
            control
                .db
                .read(|c| c.query_row("SELECT n FROM revision", [], |r| r.get::<_, i64>(0)))
                .await
                .unwrap()
        };
        let before = revision().await;
        control.ensure_owner(owner(1)).await.unwrap();
        control.note_version(LibraryRef::User(1), 2).await.unwrap();
        control
            .set_meta(vec![("schema", "{}".into())])
            .await
            .unwrap();
        assert_eq!(revision().await, before + 3);
        // Reading changes nothing.
        control.user(1).await.unwrap();
        assert_eq!(revision().await, before + 3);
    }
}