defter 0.6.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 userID.
    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>,
}

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";

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)?,
    })
}

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, created_at)
         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)",
        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,
            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 signing in with an ORCID iD 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,
}

/// 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>(8)?).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 users and 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.
const ID_SPACE: u64 = 100_000;

/// A random ID for a new user or group: in neither `table` nor, when `dir` is
/// known, a library file there (which a control database restored from a
/// backup may not know about).
fn free_id(
    conn: &Connection,
    table: &str,
    dir: Option<&Path>,
    library: fn(u64) -> LibraryRef,
) -> 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) % ID_SPACE + 1;
        let used: bool = conn.query_row(
            &format!("SELECT count(*) > 0 FROM {table} WHERE id = ?1"),
            [id as i64],
            |r| r.get(0),
        )?;
        let file = dir.is_some_and(|dir| crate::registry::path(dir, library(id)).exists());
        if !used && !file {
            return Ok(id as i64);
        }
    }
    Err(rusqlite::Error::SqliteFailure(
        rusqlite::ffi::Error::new(rusqlite::ffi::SQLITE_FULL),
        Some(format!("no free {table} ID left")),
    ))
}

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 a new user if invited or
    /// an admin. An admin's iD becomes the owner's when the owner has none yet,
    /// so the owner of a server from before never gets a second, empty library.
    pub async fn sign_in(&self, orcid: &str, admin: bool, owner: u64) -> Result<SignIn, AppError> {
        let orcid = orcid.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)
                    });
                }
                if admin
                    && let Some(mut owner) = user(&tx, owner)?
                    && owner.orcid.is_none()
                {
                    tx.execute(
                        "UPDATE user SET orcid = ?2 WHERE id = ?1",
                        params![owner.id as i64, orcid],
                    )?;
                    tx.commit()?;
                    owner.orcid = Some(orcid);
                    return Ok(if owner.disabled {
                        SignIn::Disabled
                    } else {
                        SignIn::User(owner)
                    });
                }
                let invited: bool = tx.query_row(
                    "SELECT count(*) > 0 FROM invite WHERE orcid = ?1 AND used_by IS NULL",
                    [&orcid],
                    |r| r.get(0),
                )?;
                if !invited && !admin {
                    return Ok(SignIn::NotInvited);
                }
                let id = free_id(&tx, "user", dir.as_deref(), LibraryRef::User)?;
                let user = User {
                    id: id as u64,
                    orcid: Some(orcid.clone()),
                    username: orcid.clone(),
                    display_name: None,
                    // Shown once they choose to.
                    profile_hidden: true,
                    disabled: false,
                    legacy_files: false,
                    quota_bytes: None,
                };
                insert_user(&tx, &user)?;
                tx.execute(
                    "UPDATE invite SET used_by = ?2, used_at = ?3 WHERE orcid = ?1",
                    params![orcid, id, now()],
                )?;
                tx.commit()?;
                Ok(SignIn::User(user))
            })
            .await
    }

    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. 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)?;
                tx.execute(
                    "UPDATE user SET disabled = ?2 WHERE id = ?1",
                    params![id as i64, disabled],
                )?;
                if disabled {
                    tx.execute(
                        "UPDATE api_key SET revoked_at = ?2
                         WHERE user_id = ?1 AND revoked_at IS NULL",
                        params![id as i64, now()],
                    )?;
                }
                tx.commit()
            })
            .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(8)?,
                        keys: r.get(9)?,
                    })
                })?;
                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_id(&tx, "grp", dir.as_deref(), LibraryRef::Group)?;
                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(8)?,
                            publications: r.get(9)?,
                            deduplicate: r.get(10)?,
                            overwrite: r.get(11)?,
                        },
                    ))
                })?;
                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,
        }
    }

    #[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 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);
    }
}