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

pub use api_types::{AppPasswordInfo, AuthMode, Mode, PimCollectionKind as PimKind, PimShareMode};
pub use pimdav::principal::UserType;
use rusqlite::{Connection, OptionalExtension, Params, params};
use webauthn_rs::prelude::Uuid;

/// Reads a text column through `parse`. The wire types are foreign to this
/// crate, so they cannot implement `FromSql`.
fn wire<T>(r: &rusqlite::Row, i: usize, parse: fn(&str) -> Option<T>) -> DbResult<T> {
    let s: String = r.get(i)?;
    parse(&s).ok_or_else(|| {
        rusqlite::Error::FromSqlConversionFailure(
            i,
            rusqlite::types::Type::Text,
            format!("unknown value {s:?}").into(),
        )
    })
}

#[derive(Debug, Clone)]
pub struct User {
    pub id: i64,
    pub name: String,
    pub is_admin: bool,
    /// Disabled users cannot sign in and their sessions are rejected.
    pub active: bool,
    /// Profile setting: single click opens entries (off = click selects).
    pub single_click: bool,
    /// Profile setting: show thumbnails in the grid.
    pub thumbnails: bool,
    /// Preferred UI language tag ("en", "de", "fr"); None = follow the
    /// browser.
    pub language: Option<String>,
    /// Profile setting: the first day of the week in the month grid, 0 for
    /// Sunday through 6 for Saturday.
    pub week_start: u8,
    /// Profile setting: the root the UI opens by default. May point at a
    /// root the user no longer has; the API filters that out.
    pub default_root_id: Option<i64>,
    /// What this account needs to sign in.
    pub auth_mode: AuthMode,
    /// Whether a password is set. False means the account signs in with
    /// passkeys only. See [`NO_PASSWORD`].
    pub has_password: bool,
}

/// The `pass_hash` of an account with no password.
///
/// A sentinel rather than NULL: `users.pass_hash` is `NOT NULL`, and SQLite
/// cannot drop that constraint without rebuilding the table. Nothing verifies
/// against it — [`Db::verify_password`] swaps in [`DUMMY_HASH`] so the reject
/// costs the same as a wrong password, and `PasswordHash::new("")` fails
/// anyway.
pub const NO_PASSWORD: &str = "";

/// How many passkeys one account may hold.
///
/// Also the exact number of credentials a named sign-in challenge lists. The
/// two are one number on purpose: the challenge pads a short list with decoys
/// so its length says nothing about the account, and that only works while no
/// account can push past the padding.
pub const PASSKEY_LIMIT: usize = 8;

/// How many app passwords one account may hold. One per client is the point.
pub const APP_PASSWORD_LIMIT: usize = 10;

/// What [`Db::delete_passkey`] did.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PasskeyDeleted {
    Gone,
    /// No such passkey, or it belongs to someone else.
    NotFound,
    /// Removing it would have left the account unreachable.
    LastCredential,
}

/// One registered passkey. `passkey` is the serialized `webauthn-rs`
/// credential; everything else is for the settings list.
#[derive(Debug, Clone)]
pub struct PasskeyRow {
    pub id: i64,
    pub name: String,
    pub created_at: String,
    pub last_used_at: Option<String>,
    pub discoverable: Option<bool>,
    /// `webauthn_rs::prelude::Passkey` as JSON.
    pub passkey: String,
}

#[derive(Debug, Clone)]
pub struct RootRow {
    pub id: i64,
    /// Path relative to the server root; "." means the whole root.
    pub path: String,
    pub mode: Mode,
}

#[derive(Debug, Clone)]
pub struct ShareRow {
    pub id: i64,
    pub token: String,
    pub creator_id: i64,
    /// Path of the shared item relative to the server root.
    pub target: String,
    pub is_file: bool,
    pub mode: Mode,
    pub created_at: String,
    pub expires_at: Option<String>,
    /// Argon2 hash of the share's password, when it has one. Resolve,
    /// listing and download all stay locked until the visitor enters it and
    /// gets an unlock cookie.
    pub password_hash: Option<String>,
}

impl ShareRow {
    pub fn is_expired(&self) -> bool {
        expired(self.expires_at.as_deref())
    }
}

/// An unparseable stamp never expires; the API refuses one on creation.
fn expired(expires_at: Option<&str>) -> bool {
    match expires_at {
        Some(e) => chrono::DateTime::parse_from_rfc3339(e)
            .map(|t| chrono::Utc::now() >= t.with_timezone(&chrono::Utc))
            .unwrap_or(false),
        None => false,
    }
}

/// A public feed of one calendar or address book.
#[derive(Debug, Clone)]
pub struct PimLink {
    pub id: i64,
    pub token: String,
    pub collection_id: i64,
    /// Events reduced to their times.
    pub busy_only: bool,
    pub created_at: String,
    pub expires_at: Option<String>,
    pub password_hash: Option<String>,
}

impl PimLink {
    pub fn is_expired(&self) -> bool {
        expired(self.expires_at.as_deref())
    }
}

/// A feed link with what the admin overview shows about it.
pub struct PimLinkWithOwner {
    pub link: PimLink,
    pub collection_name: String,
    pub kind: PimKind,
    pub owner_id: i64,
    pub owner_name: String,
    pub owner_active: bool,
}

/// A [`ShareRow`] together with the account that created it.
#[derive(Debug, Clone)]
pub struct ShareWithCreator {
    pub share: ShareRow,
    pub creator_name: String,
    /// Whether that account can still sign in. Deactivating an account leaves
    /// its shares live.
    pub creator_active: bool,
}

fn pim_kind(s: &str) -> Option<PimKind> {
    match s {
        "cal" => Some(PimKind::Calendar),
        "card" => Some(PimKind::Addressbook),
        _ => None,
    }
}

fn pim_kind_str(kind: PimKind) -> &'static str {
    match kind {
        PimKind::Calendar => "cal",
        PimKind::Addressbook => "card",
    }
}

/// A calendar or address book.
#[derive(Debug, Clone, Default)]
pub struct PimCollection {
    pub id: i64,
    /// The URL segment.
    pub slug: String,
    pub displayname: Option<String>,
    pub description: Option<String>,
    pub color: Option<String>,
    /// A VCALENDAR holding one VTIMEZONE.
    pub timezone: Option<String>,
    pub sort_order: Option<String>,
    /// The component types a calendar takes, comma-separated. Empty for an
    /// address book.
    pub components: String,
    /// Adds no busy time to scheduling.
    pub transparent: bool,
    /// Grows with every change to the collection or its members.
    pub seq: i64,
}

/// Where client properties the server does not interpret are kept.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PropPlace {
    Principal(i64),
    Home(i64, PimKind),
    Collection(i64),
}

impl PropPlace {
    fn key(self) -> String {
        match self {
            PropPlace::Principal(id) => format!("principal:{id}"),
            PropPlace::Home(id, kind) => format!("home-{}:{id}", pim_kind_str(kind)),
            PropPlace::Collection(id) => format!("collection:{id}"),
        }
    }

    /// The principal and collection the rows cascade with.
    fn owners(self) -> (Option<i64>, Option<i64>) {
        match self {
            PropPlace::Principal(id) | PropPlace::Home(id, _) => (Some(id), None),
            PropPlace::Collection(id) => (None, Some(id)),
        }
    }
}

/// A client property: its name and the whole property element as XML.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DeadProp {
    pub ns: String,
    pub name: String,
    pub xml: String,
}

/// A calendar or address object, without its data.
#[derive(Debug, Clone, Default)]
pub struct PimObject {
    /// The URL segment.
    pub name: String,
    pub uid: String,
    /// `VEVENT`, `VTODO`, `VJOURNAL` or `VCARD`.
    pub component: String,
    /// With the quotes.
    pub etag: String,
    pub size: i64,
    pub modified_at: String,
    /// With the quotes. Only on scheduling objects.
    pub schedule_tag: Option<String>,
}

/// One write of [`Db::pim_apply`].
#[derive(Debug)]
pub enum PimOp {
    Put {
        collection_id: i64,
        obj: PimObject,
        data: Vec<u8>,
    },
    Delete {
        collection_id: i64,
        name: String,
    },
    /// A scheduling message for the inbox of `principal_id`. `obj.uid` must
    /// be unique in the inbox: several messages share one iCalendar UID.
    Inbox {
        principal_id: i64,
        obj: PimObject,
        data: Vec<u8>,
    },
    DeleteCollection(i64),
}

#[derive(Debug, PartialEq, Eq)]
pub enum PimWrite {
    Created,
    Updated,
    NotFound,
    PreconditionFailed,
    /// Another object in the collection has this UID, under this name.
    UidConflict(String),
}

/// The `If-Match` and `If-None-Match` headers of a write.
#[derive(Debug, Default)]
pub struct Precondition {
    pub if_match: Option<String>,
    pub if_none_match: Option<String>,
}

impl Precondition {
    /// Whether the write may go ahead given the current ETag, if any.
    pub fn allows(&self, current: Option<&str>) -> bool {
        // If-Match compares strongly, If-None-Match weakly (RFC 9110, 13.1).
        let listed = |header: &str, weak: bool| match current {
            Some(etag) => header.split(',').any(|t| {
                let t = t.trim();
                let t = if weak {
                    t.strip_prefix("W/").unwrap_or(t)
                } else {
                    t
                };
                t == "*" || t == etag
            }),
            None => false,
        };
        self.if_match.as_deref().is_none_or(|h| listed(h, false))
            && !self
                .if_none_match
                .as_deref()
                .is_some_and(|h| listed(h, true))
    }
}

/// The columns [`map_user`] reads, in order. Qualified by the table name,
/// not an alias, so `RETURNING` can use them too.
const USER_COLS: &str = "users.id, users.name, users.is_admin != 0, users.active != 0,
     users.single_click != 0, users.thumbnails != 0, users.language, users.default_root_id,
     users.auth_mode, users.pass_hash != '', users.week_start";
/// How many columns [`USER_COLS`] covers. The joined queries read the root
/// columns starting here.
const USER_COL_COUNT: usize = 11;

/// Every query can fail, and every caller decides what to do about it.
///
/// Earlier versions swallowed read errors and returned a default (an empty
/// root list, a count of 0). That turned a broken database into a plausible
/// answer: "you have no folders" instead of an error. One contract now.
pub type DbResult<T> = Result<T, rusqlite::Error>;

#[derive(Clone)]
pub struct Db {
    conn: Arc<tokio::sync::Mutex<Connection>>,
    /// Principals whose default collections exist. Saves the three writes
    /// of [`Db::pim_ensure_defaults`] on every DAV request.
    defaults: Arc<std::sync::Mutex<std::collections::HashSet<i64>>>,
}

impl Db {
    pub async fn open(path: &Path) -> anyhow::Result<Self> {
        if let Some(parent) = path.parent()
            && !parent.as_os_str().is_empty()
        {
            std::fs::create_dir_all(parent)?;
        }
        let conn = Connection::open(path)?;
        conn.pragma_update(None, "journal_mode", "WAL")?;
        // WAL plus NORMAL: fsync only at checkpoints. A crash can lose the
        // last commits, never the database file.
        conn.pragma_update(None, "synchronous", "NORMAL")?;
        conn.pragma_update(None, "foreign_keys", "ON")?;
        conn.pragma_update(None, "busy_timeout", "5000")?;
        Self::migrate(&conn)?;
        Ok(Self {
            conn: Arc::new(tokio::sync::Mutex::new(conn)),
            defaults: Default::default(),
        })
    }

    fn migrate(conn: &Connection) -> rusqlite::Result<()> {
        conn.execute(
            "CREATE TABLE IF NOT EXISTS meta (key TEXT PRIMARY KEY, value TEXT NOT NULL)",
            [],
        )?;
        let version: usize = conn
            .query_row(
                "SELECT value FROM meta WHERE key = 'schema_version'",
                [],
                |r| r.get::<_, String>(0),
            )
            .optional()?
            .and_then(|v| v.parse().ok())
            .unwrap_or(0);

        let tx = conn.unchecked_transaction()?;
        for sql in MIGRATIONS.iter().skip(version) {
            conn.execute_batch(sql)?;
        }
        conn.execute(
            "INSERT OR REPLACE INTO meta (key, value) VALUES ('schema_version', ?1)",
            [MIGRATIONS.len().to_string()],
        )?;
        tx.commit()
    }

    async fn exec(&self, sql: &str, p: impl Params + Send) -> DbResult<usize> {
        self.conn.lock().await.prepare_cached(sql)?.execute(p)
    }

    async fn row<T>(
        &self,
        sql: &str,
        p: impl Params + Send,
        f: impl FnOnce(&rusqlite::Row) -> DbResult<T> + Send,
    ) -> DbResult<Option<T>> {
        self.conn
            .lock()
            .await
            .prepare_cached(sql)?
            .query_row(p, f)
            .optional()
    }

    async fn rows<T>(
        &self,
        sql: &str,
        p: impl Params + Send,
        f: impl FnMut(&rusqlite::Row) -> DbResult<T> + Send,
    ) -> DbResult<Vec<T>> {
        self.conn
            .lock()
            .await
            .prepare_cached(sql)?
            .query_map(p, f)?
            .collect()
    }

    // ---------- users ----------

    pub async fn user_count(&self) -> DbResult<i64> {
        Ok(self
            .row("SELECT COUNT(*) FROM users", [], |r| r.get(0))
            .await?
            .unwrap_or(0))
    }

    /// Create the first admin account with the whole root visible (read-write).
    ///
    /// `None` means a user already existed. The `WHERE NOT EXISTS` guard runs
    /// inside the same transaction as the insert, so two concurrent first-boot
    /// setups cannot both win; a caller's earlier `user_count` check is only
    /// an optimization, not the guarantee.
    pub async fn create_admin(&self, name: &str, pass_hash: &str) -> DbResult<Option<User>> {
        let mut c = self.conn.lock().await;
        let tx = c.transaction()?;
        let user = tx
            .query_row(
                &format!(
                    "INSERT INTO users (name, pass_hash, is_admin, created_at)
                     SELECT ?1, ?2, 1, ?3 WHERE NOT EXISTS (SELECT 1 FROM users)
                     RETURNING {USER_COLS}"
                ),
                params![name, pass_hash, now()],
                map_user,
            )
            .optional()?;
        let Some(user) = user else {
            return Ok(None); // dropping `tx` rolls back
        };
        insert_roots(&tx, user.id, &[(".".to_string(), Mode::Rw)])?;
        tx.commit()?;
        Ok(Some(user))
    }

    pub async fn verify_password(&self, name: &str, password: &str) -> DbResult<Option<User>> {
        // The guard is scoped to the query alone. Argon2 below is slow by
        // design; holding the single connection lock across it would make one
        // login serialize every other database access.
        let row: Option<(User, String)> = self
            .row(
                &format!("SELECT {USER_COLS}, pass_hash FROM users WHERE name = ?1"),
                [name],
                |r| Ok((map_user(r)?, r.get(USER_COL_COUNT)?)),
            )
            .await?;
        // An unknown name, a disabled account and a passkey-only account all
        // still pay for one Argon2 verify, so the response time does not tell
        // them apart from a real account with a wrong password.
        let (user, hash) = match row {
            Some((u, h)) if u.active && u.has_password => (Some(u), h),
            _ => (None, DUMMY_HASH.clone()),
        };
        let ok = crate::auth::verify_password_async(password, &hash).await;
        Ok(user.filter(|_| ok))
    }

    pub async fn create_session(&self, user_id: i64, token: &str) -> DbResult<()> {
        self.exec(
            "INSERT INTO sessions (token, user_id, created_at, last_seen_at)
             VALUES (?1, ?2, ?3, ?4)",
            (token, user_id, now(), now()),
        )
        .await?;
        Ok(())
    }

    pub async fn delete_session(&self, token: &str) -> DbResult<()> {
        self.exec("DELETE FROM sessions WHERE token = ?1", [token])
            .await?;
        Ok(())
    }

    /// The session's user plus that user's roots, in one round trip. Every
    /// authenticated request needs both, so they are not two queries.
    pub async fn session_user_with_roots(
        &self,
        token: &str,
    ) -> DbResult<Option<(User, Vec<RootRow>)>> {
        let c = self.conn.lock().await;
        let mut stmt = c.prepare_cached(&format!(
            "SELECT {USER_COLS}, r.id, r.path, r.mode
             FROM sessions s
             JOIN users ON users.id = s.user_id
             LEFT JOIN user_roots r ON r.user_id = users.id
             WHERE s.token = ?1 AND users.active = 1
             ORDER BY r.id",
        ))?;
        // One row per root; a user without roots still returns one row, with
        // the root columns NULL.
        let mut user: Option<User> = None;
        let mut roots: Vec<RootRow> = Vec::new();
        let mut rows = stmt.query([token])?;
        while let Some(r) = rows.next()? {
            if user.is_none() {
                user = Some(map_user(r)?);
            }
            roots.extend(map_root_at(r, USER_COL_COUNT)?);
        }
        Ok(user.map(|u| (u, roots)))
    }

    // ---------- roots ----------

    pub async fn user_roots(&self, user_id: i64) -> DbResult<Vec<RootRow>> {
        let roots = self
            .rows(
                "SELECT id, path, mode FROM user_roots WHERE user_id = ?1 ORDER BY id",
                [user_id],
                |r| map_root_at(r, 0),
            )
            .await?;
        Ok(roots.into_iter().flatten().collect())
    }

    // ---------- admin: user management (M7) ----------

    /// Every user with their roots, in one query. The admin user list needs
    /// both, and a per-user roots query would be one round trip per user.
    pub async fn all_users_with_roots(&self) -> DbResult<Vec<(User, Vec<RootRow>)>> {
        let c = self.conn.lock().await;
        let mut stmt = c.prepare_cached(&format!(
            "SELECT {USER_COLS}, r.id, r.path, r.mode
             FROM users
             LEFT JOIN user_roots r ON r.user_id = users.id
             ORDER BY users.id, r.id",
        ))?;
        // Rows arrive grouped by user, so a new user id starts a new group.
        let mut out: Vec<(User, Vec<RootRow>)> = Vec::new();
        let mut rows = stmt.query([])?;
        while let Some(r) = rows.next()? {
            let uid: i64 = r.get(0)?;
            if out.last().is_none_or(|(u, _)| u.id != uid) {
                out.push((map_user(r)?, Vec::new()));
            }
            let roots = &mut out.last_mut().expect("pushed above").1;
            roots.extend(map_root_at(r, USER_COL_COUNT)?);
        }
        Ok(out)
    }

    pub async fn find_user_by_id(&self, id: i64) -> DbResult<Option<User>> {
        self.row(
            &format!("SELECT {USER_COLS} FROM users WHERE id = ?1"),
            [id],
            map_user,
        )
        .await
    }

    pub async fn find_user_by_name(&self, name: &str) -> DbResult<Option<User>> {
        self.row(
            &format!("SELECT {USER_COLS} FROM users WHERE name = ?1"),
            [name],
            map_user,
        )
        .await
    }

    pub async fn count_admins(&self) -> DbResult<i64> {
        Ok(self
            .row(
                "SELECT COUNT(*) FROM users WHERE is_admin = 1 AND active = 1",
                [],
                |r| r.get(0),
            )
            .await?
            .unwrap_or(0))
    }

    /// Create a user with the given roots (path, mode) pairs.
    pub async fn create_user(
        &self,
        name: &str,
        pass_hash: &str,
        is_admin: bool,
        roots: &[(String, Mode)],
    ) -> DbResult<User> {
        let mut c = self.conn.lock().await;
        let tx = c.transaction()?;
        let user = tx.query_row(
            &format!(
                "INSERT INTO users (name, pass_hash, is_admin, active, created_at)
                 VALUES (?1, ?2, ?3, 1, ?4) RETURNING {USER_COLS}"
            ),
            params![name, pass_hash, is_admin as i64, now()],
            map_user,
        )?;
        insert_roots(&tx, user.id, roots)?;
        tx.commit()?;
        Ok(user)
    }

    /// Write the profile settings a user edits for themselves.
    pub async fn set_user_profile(&self, u: &User) -> DbResult<()> {
        self.exec(
            "UPDATE users SET single_click = ?1, thumbnails = ?2, language = ?3,
                              default_root_id = ?4, week_start = ?5
             WHERE id = ?6",
            (
                u.single_click,
                u.thumbnails,
                u.language.as_deref(),
                u.default_root_id,
                u.week_start,
                u.id,
            ),
        )
        .await?;
        Ok(())
    }

    /// Apply an admin edit atomically: every `Some` field is written in one
    /// transaction, so a failure midway leaves the user unchanged.
    pub async fn update_user(
        &self,
        id: i64,
        pass_hash: Option<&str>,
        is_admin: Option<bool>,
        active: Option<bool>,
        roots: Option<&[(String, Mode)]>,
    ) -> DbResult<()> {
        let mut c = self.conn.lock().await;
        let tx = c.transaction()?;
        // An admin sets a password to get someone back into a locked-out
        // account, so every other way in goes with it: the passkeys, the app
        // passwords, the requirement to use one. Always rather than on
        // request, so a reset cannot leave a credential the user no longer
        // has.
        if let Some(h) = pass_hash {
            set_password(&tx, id, h)?;
            tx.execute("DELETE FROM passkeys WHERE user_id = ?1", [id])?;
            tx.execute("DELETE FROM app_passwords WHERE user_id = ?1", [id])?;
            tx.execute(
                "UPDATE users SET auth_mode = ?1 WHERE id = ?2",
                params![AuthMode::Either.as_str(), id],
            )?;
        }
        if let Some(a) = is_admin {
            tx.execute(
                "UPDATE users SET is_admin = ?1 WHERE id = ?2",
                params![a as i64, id],
            )?;
        }
        if let Some(a) = active {
            tx.execute(
                "UPDATE users SET active = ?1 WHERE id = ?2",
                params![a as i64, id],
            )?;
        }
        if let Some(roots) = roots {
            tx.execute("DELETE FROM user_roots WHERE user_id = ?1", [id])?;
            insert_roots(&tx, id, roots)?;
        }
        tx.commit()
    }

    // ---------- credentials: password, passkeys, sign-in mode ----------

    /// Set the password. Leaves every session alone, so the caller must pair
    /// this with [`Db::delete_other_sessions`] — see
    /// [`crate::api::passkeys`]'s `invalidate_elsewhere`. The admin path uses
    /// [`set_password`] instead, which ends every session including its own.
    ///
    /// Only for setting a real one. Clearing it is [`Db::clear_user_password`],
    /// which has a rule to keep.
    pub async fn set_password_keeping_sessions(&self, id: i64, pass_hash: &str) -> DbResult<()> {
        self.exec(
            "UPDATE users SET pass_hash = ?1 WHERE id = ?2",
            (pass_hash, id),
        )
        .await?;
        Ok(())
    }

    /// Leave the account on its passkeys alone. `false` means that would have
    /// locked it out, so nothing changed.
    pub async fn clear_user_password(&self, id: i64) -> DbResult<bool> {
        let mut c = self.conn.lock().await;
        let tx = c.transaction()?;
        tx.execute(
            "UPDATE users SET pass_hash = ?1 WHERE id = ?2",
            params![NO_PASSWORD, id],
        )?;
        commit_if_reachable(tx, id)
    }

    /// `false` means the account does not satisfy the new mode, so it stands.
    pub async fn set_user_auth_mode(&self, id: i64, mode: AuthMode) -> DbResult<bool> {
        let mut c = self.conn.lock().await;
        let tx = c.transaction()?;
        tx.execute(
            "UPDATE users SET auth_mode = ?1 WHERE id = ?2",
            params![mode.as_str(), id],
        )?;
        commit_if_reachable(tx, id)
    }

    /// Drop every session of this user except `keep`.
    ///
    /// Called after any credential change. Otherwise a session stolen before
    /// the change keeps working for its full 30 days.
    pub async fn delete_other_sessions(&self, user_id: i64, keep: &str) -> DbResult<()> {
        self.exec(
            "DELETE FROM sessions WHERE user_id = ?1 AND token != ?2",
            (user_id, keep),
        )
        .await?;
        Ok(())
    }

    /// This account's WebAuthn user handle, creating it on first use.
    ///
    /// Stable for the lifetime of the account: a discoverable passkey stores
    /// this value and hands it back at sign-in, so changing it would orphan
    /// every existing passkey.
    pub async fn user_webauthn_id(&self, id: i64) -> DbResult<Uuid> {
        let c = self.conn.lock().await;
        let existing: Option<String> = c
            .query_row("SELECT webauthn_id FROM users WHERE id = ?1", [id], |r| {
                r.get(0)
            })
            .optional()?
            .flatten();
        if let Some(parsed) = existing.as_deref().and_then(|s| Uuid::parse_str(s).ok()) {
            return Ok(parsed);
        }
        let fresh = Uuid::new_v4();
        c.execute(
            "UPDATE users SET webauthn_id = ?1 WHERE id = ?2",
            params![fresh.to_string(), id],
        )?;
        Ok(fresh)
    }

    /// The account a discoverable credential's user handle points at.
    pub async fn find_user_by_webauthn_id(&self, wid: &Uuid) -> DbResult<Option<User>> {
        self.row(
            &format!("SELECT {USER_COLS} FROM users WHERE webauthn_id = ?1"),
            [wid.to_string()],
            map_user,
        )
        .await
    }

    pub async fn user_passkeys(&self, user_id: i64) -> DbResult<Vec<PasskeyRow>> {
        self.rows(
            &format!("SELECT {PASSKEY_COLS} FROM passkeys WHERE user_id = ?1 ORDER BY id"),
            [user_id],
            map_passkey,
        )
        .await
    }

    pub async fn count_passkeys(&self, user_id: i64) -> DbResult<i64> {
        Ok(self
            .row(
                "SELECT COUNT(*) FROM passkeys WHERE user_id = ?1",
                [user_id],
                |r| r.get(0),
            )
            .await?
            .unwrap_or(0))
    }

    /// The per-install secret behind the decoy credentials a named passkey
    /// challenge is padded with. Created on first use, so no migration.
    pub async fn decoy_secret(&self) -> DbResult<String> {
        let c = self.conn.lock().await;
        let existing: Option<String> = c
            .query_row(
                "SELECT value FROM meta WHERE key = 'decoy_secret'",
                [],
                |r| r.get(0),
            )
            .optional()?;
        if let Some(secret) = existing {
            return Ok(secret);
        }
        let fresh = crate::auth::random_token();
        c.execute(
            "INSERT OR REPLACE INTO meta (key, value) VALUES ('decoy_secret', ?1)",
            [&fresh],
        )?;
        Ok(fresh)
    }

    /// The length in bytes of every credential id registered on this server,
    /// one entry per credential.
    ///
    /// Decoys draw their length from this list. Authenticators disagree on it —
    /// a security key emits 64 bytes, a platform key often 16 or 32 — so a
    /// decoy of one fixed length would stand out next to the real entries.
    ///
    /// Duplicates are kept deliberately. Drawing from the distinct lengths
    /// would make a rare length as likely as a common one, and decoys that do
    /// not match how the install actually looks are the thing worth avoiding.
    pub async fn cred_id_lengths(&self) -> DbResult<Vec<usize>> {
        self.rows(
            "SELECT length(cred_id) FROM passkeys ORDER BY id",
            [],
            |r| Ok(r.get::<_, i64>(0)?.max(1) as usize),
        )
        .await
    }

    /// Store a freshly registered passkey. A duplicate `cred_id` is a unique
    /// violation, which is the intended answer: the same credential must not
    /// be registered twice, not even to a second account.
    ///
    /// `None` means the account is already at [`PASSKEY_LIMIT`]. The count and
    /// the insert share one transaction, so two registrations landing together
    /// cannot put the account one over and give its sign-in challenge a
    /// telltale length.
    pub async fn add_passkey(
        &self,
        user_id: i64,
        cred_id: &[u8],
        passkey: &str,
        name: &str,
        discoverable: Option<bool>,
    ) -> DbResult<Option<PasskeyRow>> {
        let mut c = self.conn.lock().await;
        let tx = c.transaction()?;
        let held: i64 = tx.query_row(
            "SELECT COUNT(*) FROM passkeys WHERE user_id = ?1",
            [user_id],
            |r| r.get(0),
        )?;
        if held as usize >= PASSKEY_LIMIT {
            return Ok(None);
        }
        let row = tx.query_row(
            &format!(
                "INSERT INTO passkeys (user_id, cred_id, passkey, name, discoverable, created_at)
                 VALUES (?1, ?2, ?3, ?4, ?5, ?6) RETURNING {PASSKEY_COLS}"
            ),
            params![user_id, cred_id, passkey, name, discoverable, now()],
            map_passkey,
        )?;
        tx.commit()?;
        Ok(Some(row))
    }

    pub async fn delete_passkey(&self, id: i64, user_id: i64) -> DbResult<PasskeyDeleted> {
        let mut c = self.conn.lock().await;
        let tx = c.transaction()?;
        let hit = tx.execute(
            "DELETE FROM passkeys WHERE id = ?1 AND user_id = ?2",
            params![id, user_id],
        )? > 0;
        if !hit {
            tx.rollback()?;
            return Ok(PasskeyDeleted::NotFound);
        }
        Ok(if commit_if_reachable(tx, user_id)? {
            PasskeyDeleted::Gone
        } else {
            PasskeyDeleted::LastCredential
        })
    }

    /// Record a successful assertion: the re-serialized credential (its
    /// signature counter and backup flags may have moved) and the time.
    pub async fn passkey_used(&self, id: i64, passkey: &str) -> DbResult<()> {
        self.exec(
            "UPDATE passkeys SET passkey = ?1, last_used_at = ?2 WHERE id = ?3",
            (passkey, now(), id),
        )
        .await?;
        Ok(())
    }

    // ---------- app passwords (WebDAV) ----------

    pub async fn app_passwords(&self, user_id: i64) -> DbResult<Vec<AppPasswordInfo>> {
        self.rows(
            &format!(
                "SELECT {APP_PASSWORD_COLS} FROM app_passwords WHERE user_id = ?1 ORDER BY id"
            ),
            [user_id],
            map_app_password,
        )
        .await
    }

    /// `None` means the account is already at [`APP_PASSWORD_LIMIT`]. Count
    /// and insert share one transaction, so the cap cannot be raced.
    pub async fn add_app_password(
        &self,
        user_id: i64,
        name: &str,
        secret_hash: &str,
    ) -> DbResult<Option<AppPasswordInfo>> {
        let mut c = self.conn.lock().await;
        let tx = c.transaction()?;
        let held: i64 = tx.query_row(
            "SELECT COUNT(*) FROM app_passwords WHERE user_id = ?1",
            [user_id],
            |r| r.get(0),
        )?;
        if held as usize >= APP_PASSWORD_LIMIT {
            return Ok(None);
        }
        let row = tx.query_row(
            &format!(
                "INSERT INTO app_passwords (user_id, name, secret_hash, created_at)
                 VALUES (?1, ?2, ?3, ?4) RETURNING {APP_PASSWORD_COLS}"
            ),
            params![user_id, name, secret_hash, now()],
            map_app_password,
        )?;
        tx.commit()?;
        Ok(Some(row))
    }

    /// `false` means no row matched: an unknown id, or someone else's.
    ///
    /// No reachability check, unlike [`Db::delete_passkey`]: an app password
    /// never signs in to the web UI.
    pub async fn delete_app_password(&self, id: i64, user_id: i64) -> DbResult<bool> {
        Ok(self
            .exec(
                "DELETE FROM app_passwords WHERE id = ?1 AND user_id = ?2",
                (id, user_id),
            )
            .await?
            > 0)
    }

    /// The account an app password opens. Stamps `last_used_at` on the way.
    ///
    /// Inactive accounts are excluded here, because nothing downstream in the
    /// WebDAV path looks at the flag.
    pub async fn user_by_app_password(&self, secret_hash: &str) -> DbResult<Option<User>> {
        let c = self.conn.lock().await;
        let user = c
            .prepare_cached(&format!(
                "SELECT {USER_COLS}
                 FROM app_passwords a
                 JOIN users ON users.id = a.user_id
                 WHERE a.secret_hash = ?1 AND users.active = 1"
            ))?
            .query_row([secret_hash], map_user)
            .optional()?;
        if user.is_some() {
            // A mount re-sends its credential on every request. An hour's
            // resolution keeps that from writing the stamp on each one.
            c.execute(
                "UPDATE app_passwords SET last_used_at = ?1
                 WHERE secret_hash = ?2 AND (last_used_at IS NULL OR last_used_at < ?3)",
                params![now(), secret_hash, stamp(chrono::Duration::hours(1))],
            )?;
        }
        Ok(user)
    }

    /// The stored spelling of the name, which a Basic login may differ from
    /// in case.
    pub async fn user_name(&self, id: i64) -> DbResult<Option<String>> {
        self.row("SELECT name FROM users WHERE id = ?1", [id], |r| r.get(0))
            .await
    }

    /// Deletes an account after `ops`, in one transaction. The ops make other
    /// principals' objects forget it (`pim_schedule::forget`). `false` means
    /// no row matched, and then no op is applied.
    pub async fn delete_user(&self, id: i64, ops: &[PimOp]) -> DbResult<bool> {
        let mut c = self.conn.lock().await;
        let tx = c.transaction()?;
        apply_ops(&tx, ops)?;
        if tx.execute("DELETE FROM users WHERE id = ?1", [id])? == 0 {
            return Ok(false);
        }
        tx.commit()?;
        self.forget_defaults();
        Ok(true)
    }

    // ---------- shares ----------

    #[allow(clippy::too_many_arguments)] // one row's columns, all required
    pub async fn create_share(
        &self,
        creator_id: i64,
        token: &str,
        target: &str,
        is_file: bool,
        mode: Mode,
        expires_at: Option<&str>,
        password_hash: Option<&str>,
    ) -> DbResult<ShareRow> {
        let c = self.conn.lock().await;
        c.query_row(
            &format!(
                "INSERT INTO shares
                   (token, creator_id, target, is_file, mode, created_at, expires_at, password_hash)
                 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8) RETURNING {SHARE_COLS}"
            ),
            params![
                token,
                creator_id,
                target,
                is_file as i64,
                mode.as_str(),
                now(),
                expires_at,
                password_hash
            ],
            map_share,
        )
    }

    /// Record that a visitor entered `share_id`'s password, and return the
    /// token that proves it (the value of their unlock cookie).
    pub async fn create_share_unlock(&self, share_id: i64) -> DbResult<String> {
        let token = crate::auth::random_token();
        self.exec(
            "INSERT INTO share_unlocks (token, share_id, created_at) VALUES (?1, ?2, ?3)",
            (&token, share_id, now()),
        )
        .await?;
        Ok(token)
    }

    /// Delete what has outlived its use. Returns how many shares and feed
    /// links, and how many unlock rows went.
    ///
    /// Housekeeping only: every read already refuses an expired share, so
    /// nothing here is load-bearing and the interval does not matter.
    pub async fn sweep(&self) -> DbResult<(usize, usize)> {
        let c = self.conn.lock().await;
        // SQLite parses the timestamp instead of comparing it as text:
        // `expires_at` is stored exactly as the client sent it and may carry
        // an offset or fractional seconds. An unparseable one yields NULL and
        // so survives, which is what `ShareRow::is_expired` decided too.
        let mut shares = 0;
        for table in ["shares", "pim_links"] {
            shares += c.execute(
                &format!(
                    "DELETE FROM {table}
                     WHERE expires_at IS NOT NULL AND julianday(expires_at) <= julianday('now')"
                ),
                [],
            )?;
        }
        // The cookie carrying an unlock is a session cookie, so it is already
        // gone from every browser. Deleting a share takes its own with it.
        let unlocks = c.execute(
            "DELETE FROM share_unlocks WHERE created_at < ?1",
            [stamp(chrono::Duration::days(UNLOCK_MAX_AGE_DAYS))],
        )?;
        drop(c);
        self.pim_prune(PIM_TOMBSTONES_KEPT).await?;
        Ok((shares, unlocks))
    }

    /// Keeps the newest `keep` deletion records of each collection.
    pub async fn pim_prune(&self, keep: i64) -> DbResult<()> {
        let c = self.conn.lock().await;
        c.execute(
            "UPDATE pim_collections SET pruned_seq = MAX(pruned_seq, old.seq)
             FROM (SELECT collection_id, MAX(seq) AS seq FROM (
                       SELECT collection_id, seq, ROW_NUMBER() OVER (
                           PARTITION BY collection_id ORDER BY seq DESC) AS rn
                       FROM pim_changes WHERE deleted != 0)
                   WHERE rn > ?1 GROUP BY collection_id) AS old
             WHERE id = old.collection_id",
            [keep],
        )?;
        c.execute(
            "DELETE FROM pim_changes WHERE deleted != 0 AND seq <= (
                 SELECT pruned_seq FROM pim_collections WHERE id = collection_id)",
            [],
        )?;
        Ok(())
    }

    /// Whether `token` is a live unlock for `share_id`.
    ///
    /// The share id is part of the lookup, so an unlock for one share cannot
    /// open another.
    pub async fn share_unlock_valid(&self, token: &str, share_id: i64) -> DbResult<bool> {
        Ok(self
            .row(
                "SELECT 1 FROM share_unlocks WHERE token = ?1 AND share_id = ?2",
                (token, share_id),
                |_| Ok(()),
            )
            .await?
            .is_some())
    }

    pub async fn share_by_token(&self, token: &str) -> DbResult<Option<ShareRow>> {
        self.row(
            &format!("SELECT {SHARE_COLS} FROM shares WHERE token = ?1"),
            [token],
            map_share,
        )
        .await
    }

    pub async fn user_shares(&self, creator_id: i64) -> DbResult<Vec<ShareRow>> {
        self.rows(
            &format!("SELECT {SHARE_COLS} FROM shares WHERE creator_id = ?1 ORDER BY id DESC"),
            [creator_id],
            map_share,
        )
        .await
    }

    /// Revoke every share on `target` or on anything beneath it. Returns how
    /// many were dropped.
    ///
    /// Called when a path stops meaning what it meant: the item was deleted,
    /// renamed, or moved away. A share names a path, and a path is not a
    /// stable identity, so leaving the row behind would let a *new* item that
    /// later takes the freed path inherit the old link's audience.
    ///
    /// `substr` rather than `LIKE`: a target containing `%` or `_` would make
    /// a `LIKE` pattern over-match and revoke unrelated shares.
    pub async fn revoke_shares_at(&self, target: &str) -> DbResult<usize> {
        self.exec(
            "DELETE FROM shares
             WHERE target = ?1 OR substr(target, 1, length(?1) + 1) = ?1 || '/'",
            [target],
        )
        .await
    }

    /// Delete a share: one of `creator`'s, or with `None` whoever created it.
    /// `false` means no row matched.
    pub async fn delete_share(&self, id: i64, creator: Option<i64>) -> DbResult<bool> {
        Ok(self
            .exec(
                "DELETE FROM shares WHERE id = ?1 AND (?2 IS NULL OR creator_id = ?2)",
                (id, creator),
            )
            .await?
            > 0)
    }

    /// Every share on the server with its creator. Grouped by account name,
    /// newest link within an account first.
    ///
    /// The join cannot miss: `shares.creator_id` cascades on delete, so a share
    /// never outlives the account that made it.
    pub async fn all_shares_with_creators(&self) -> DbResult<Vec<ShareWithCreator>> {
        self.rows(
            &format!(
                "SELECT {SHARE_COLS}, u.name, u.active != 0
                 FROM shares
                 JOIN users u ON u.id = shares.creator_id
                 ORDER BY u.name COLLATE NOCASE, shares.id DESC"
            ),
            [],
            |r| {
                Ok(ShareWithCreator {
                    share: map_share(r)?,
                    creator_name: r.get(9)?,
                    creator_active: r.get(10)?,
                })
            },
        )
        .await
    }

    // ---------- settings ----------

    /// Folders excluded from search, as paths relative to the server root.
    ///
    /// Stored as one JSON array in a settings row. A table of its own would
    /// be overkill for a hand-edited list read once per search.
    pub async fn search_excludes(&self) -> DbResult<Vec<String>> {
        let raw = self.get_setting("search_excludes").await?;
        // Normalised on read as well as on write. A value edited straight
        // into the database would otherwise never match: `is_excluded`
        // compares against paths with no slash at either end.
        let clean = |v: Vec<String>| -> Vec<String> {
            v.into_iter()
                .map(|p| p.trim().replace('\\', "/").trim_matches('/').to_string())
                .filter(|p| !p.is_empty() && p != ".")
                .collect()
        };
        // A hand-edited, unparseable value falls back to no exclusions,
        // the same as an absent row.
        Ok(raw
            .as_deref()
            .and_then(|v| serde_json::from_str::<Vec<String>>(v).ok())
            .map(clean)
            .unwrap_or_default())
    }

    pub async fn set_search_excludes(&self, paths: &[String]) -> DbResult<()> {
        let json = serde_json::to_string(paths).unwrap_or_else(|_| "[]".to_string());
        self.set_setting("search_excludes", &json).await
    }

    pub async fn get_setting(&self, key: &str) -> DbResult<Option<String>> {
        self.row("SELECT value FROM settings WHERE key = ?1", [key], |r| {
            r.get(0)
        })
        .await
    }

    pub async fn set_setting(&self, key: &str, value: &str) -> DbResult<()> {
        self.exec(
            "INSERT INTO settings (key, value) VALUES (?1, ?2)
             ON CONFLICT(key) DO UPDATE SET value = ?2",
            (key, value),
        )
        .await?;
        Ok(())
    }

    // ---------- CalDAV and CardDAV ----------

    /// Gives an account's principal a calendar, an address book and a
    /// scheduling inbox when it has none.
    pub async fn pim_ensure_defaults(&self, principal_id: i64) -> DbResult<()> {
        if self.defaults_known(principal_id) {
            return Ok(());
        }
        let c = self.conn.lock().await;
        let mut stmt = c.prepare_cached(
            "INSERT INTO pim_collections (principal_id, kind, slug, displayname, components, created_at)
             SELECT ?1, ?2, 'default', ?3, ?4, ?5
             WHERE NOT EXISTS (SELECT 1 FROM pim_collections
                               WHERE principal_id = ?1 AND kind = ?2 AND slug != 'inbox')",
        )?;
        let now = now();
        stmt.execute(params![
            principal_id,
            "cal",
            "Calendar",
            "VEVENT,VTODO,VJOURNAL",
            now
        ])?;
        stmt.execute(params![principal_id, "card", "Contacts", "", now])?;
        ensure_inbox(&c, principal_id)?;
        self.defaults_lock().insert(principal_id);
        Ok(())
    }

    fn defaults_lock(&self) -> std::sync::MutexGuard<'_, std::collections::HashSet<i64>> {
        self.defaults.lock().unwrap_or_else(|e| e.into_inner())
    }

    fn defaults_known(&self, principal_id: i64) -> bool {
        self.defaults_lock().contains(&principal_id)
    }

    /// After a collection or principal is deleted, defaults may be missing
    /// again, and a principal id may be reused.
    fn forget_defaults(&self) {
        self.defaults_lock().clear();
    }

    /// The scheduling inbox alone, for rooms and resources.
    pub async fn pim_ensure_inbox(&self, principal_id: i64) -> DbResult<()> {
        let c = self.conn.lock().await;
        ensure_inbox(&c, principal_id)
    }

    /// The calendar that receives new invitations of `component`: the one
    /// the principal chose, else the oldest one that takes it.
    pub async fn pim_calendar_for(
        &self,
        principal_id: i64,
        component: &str,
    ) -> DbResult<Option<PimCollection>> {
        self.row(
            &format!(
                "SELECT {PIM_COLLECTION_COLS} FROM pim_collections c
                 WHERE c.principal_id = ?1 AND c.kind = 'cal' AND c.slug != 'inbox'
                     AND ',' || c.components || ',' LIKE '%,' || ?2 || ',%'
                 ORDER BY c.id IS NOT (SELECT default_calendar_id FROM principals WHERE id = ?1),
                     c.id
                 LIMIT 1"
            ),
            (principal_id, component),
            map_pim_collection,
        )
        .await
    }

    /// The calendar [`Db::pim_calendar_for`] prefers. `None`: the oldest.
    pub async fn pim_set_default_calendar(
        &self,
        principal_id: i64,
        collection_id: Option<i64>,
    ) -> DbResult<()> {
        self.exec(
            "UPDATE principals SET default_calendar_id = ?2 WHERE id = ?1",
            (principal_id, collection_id),
        )
        .await?;
        Ok(())
    }

    /// The object with `uid` in any of the user's own calendars.
    pub async fn pim_find_uid(
        &self,
        principal_id: i64,
        uid: &str,
    ) -> DbResult<Option<(i64, PimObject, Vec<u8>)>> {
        self.row(
            &format!(
                "SELECT {PIM_OBJECT_COLS}, o.data, o.collection_id
                 FROM pim_objects o JOIN pim_collections c ON c.id = o.collection_id
                 WHERE c.principal_id = ?1 AND c.kind = 'cal' AND c.slug != 'inbox'
                     AND o.uid = ?2
                 ORDER BY o.schedule_tag IS NULL, o.id LIMIT 1"
            ),
            (principal_id, uid),
            |r| Ok((r.get(8)?, map_pim_object(r)?, r.get(7)?)),
        )
        .await
    }

    /// The name of another object in the collection that has `uid`.
    pub async fn pim_uid_holder(
        &self,
        collection_id: i64,
        uid: &str,
        name: &str,
    ) -> DbResult<Option<String>> {
        self.row(
            "SELECT name FROM pim_objects WHERE collection_id = ?1 AND uid = ?2 AND name != ?3",
            (collection_id, uid, name),
            |r| r.get(0),
        )
        .await
    }

    /// Several object writes in one transaction.
    pub async fn pim_apply(&self, ops: &[PimOp]) -> DbResult<()> {
        let mut c = self.conn.lock().await;
        let tx = c.transaction()?;
        apply_ops(&tx, ops)?;
        tx.commit()?;
        if ops.iter().any(|o| matches!(o, PimOp::DeleteCollection(_))) {
            self.forget_defaults();
        }
        Ok(())
    }

    pub async fn pim_collections(
        &self,
        principal_id: i64,
        kind: PimKind,
    ) -> DbResult<Vec<PimCollection>> {
        self.rows(
            &format!(
                "SELECT {PIM_COLLECTION_COLS} FROM pim_collections c
                 WHERE c.principal_id = ?1 AND c.kind = ?2 ORDER BY c.id"
            ),
            (principal_id, pim_kind_str(kind)),
            map_pim_collection,
        )
        .await
    }

    pub async fn pim_collection(
        &self,
        principal_id: i64,
        kind: PimKind,
        slug: &str,
    ) -> DbResult<Option<PimCollection>> {
        self.row(
            &format!(
                "SELECT {PIM_COLLECTION_COLS} FROM pim_collections c
                 WHERE c.principal_id = ?1 AND c.kind = ?2 AND c.slug = ?3"
            ),
            (principal_id, pim_kind_str(kind), slug),
            map_pim_collection,
        )
        .await
    }

    /// `false` if the slug is taken. `id` and `seq` of `new` are ignored.
    pub async fn pim_create_collection(
        &self,
        principal_id: i64,
        kind: PimKind,
        new: &PimCollection,
        props: &[DeadProp],
    ) -> DbResult<bool> {
        let mut c = self.conn.lock().await;
        let tx = c.transaction()?;
        let n = tx.execute(
            "INSERT INTO pim_collections (principal_id, kind, slug, displayname,
                 description, color, timezone, sort_order, components, transparent, created_at)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)
             ON CONFLICT (principal_id, kind, slug) DO NOTHING",
            params![
                principal_id,
                pim_kind_str(kind),
                new.slug,
                new.displayname,
                new.description,
                new.color,
                new.timezone,
                new.sort_order,
                new.components,
                new.transparent,
                now()
            ],
        )?;
        if n == 0 {
            return Ok(false);
        }
        let place = PropPlace::Collection(tx.last_insert_rowid());
        write_props(&tx, place, props, &[])?;
        tx.commit()?;
        Ok(true)
    }

    /// Writes the properties `col` changes from `before` to `after`, if
    /// given, and the client properties of `place`, in one transaction. A
    /// collection counts it as a change. Unchanged columns keep what a
    /// concurrent writer stored.
    pub async fn pim_patch(
        &self,
        place: PropPlace,
        col: Option<(&PimCollection, &PimCollection)>,
        set: &[DeadProp],
        remove: &[(String, String)],
    ) -> DbResult<()> {
        let mut c = self.conn.lock().await;
        let tx = c.transaction()?;
        if let Some((before, after)) = col {
            let columns: [(&str, bool, &dyn rusqlite::ToSql); 6] = [
                (
                    "displayname",
                    before.displayname != after.displayname,
                    &after.displayname,
                ),
                (
                    "description",
                    before.description != after.description,
                    &after.description,
                ),
                ("color", before.color != after.color, &after.color),
                (
                    "timezone",
                    before.timezone != after.timezone,
                    &after.timezone,
                ),
                (
                    "sort_order",
                    before.sort_order != after.sort_order,
                    &after.sort_order,
                ),
                (
                    "transparent",
                    before.transparent != after.transparent,
                    &after.transparent,
                ),
            ];
            let mut sql = "UPDATE pim_collections SET seq = seq + 1".to_string();
            let mut values: Vec<&dyn rusqlite::ToSql> = vec![&after.id];
            for (name, changed, value) in columns {
                if changed {
                    values.push(value);
                    sql.push_str(&format!(", {name} = ?{}", values.len()));
                }
            }
            sql.push_str(" WHERE id = ?1");
            tx.execute(&sql, values.as_slice())?;
        }
        write_props(&tx, place, set, remove)?;
        tx.commit()?;
        Ok(())
    }

    pub async fn pim_props(&self, place: PropPlace) -> DbResult<Vec<DeadProp>> {
        self.rows(
            "SELECT ns, name, xml FROM pim_props WHERE place = ?1 ORDER BY rowid",
            [place.key()],
            |r| {
                Ok(DeadProp {
                    ns: r.get(0)?,
                    name: r.get(1)?,
                    xml: r.get(2)?,
                })
            },
        )
        .await
    }

    pub async fn pim_objects(&self, collection_id: i64) -> DbResult<Vec<PimObject>> {
        self.rows(
            &format!(
                "SELECT {PIM_OBJECT_COLS} FROM pim_objects o
                 WHERE o.collection_id = ?1 ORDER BY o.name"
            ),
            [collection_id],
            map_pim_object,
        )
        .await
    }

    pub async fn pim_object(
        &self,
        collection_id: i64,
        name: &str,
    ) -> DbResult<Option<(PimObject, Vec<u8>)>> {
        self.row(
            &format!(
                "SELECT {PIM_OBJECT_COLS}, o.data FROM pim_objects o
                 WHERE o.collection_id = ?1 AND o.name = ?2"
            ),
            (collection_id, name),
            |r| Ok((map_pim_object(r)?, r.get(7)?)),
        )
        .await
    }

    pub async fn pim_objects_with_data(
        &self,
        collection_id: i64,
    ) -> DbResult<Vec<(PimObject, Vec<u8>)>> {
        self.rows(
            &format!(
                "SELECT {PIM_OBJECT_COLS}, o.data FROM pim_objects o
                 WHERE o.collection_id = ?1 ORDER BY o.name"
            ),
            [collection_id],
            |r| Ok((map_pim_object(r)?, r.get(7)?)),
        )
        .await
    }

    /// `(name, seq, deleted)` of the members changed after `since`, oldest
    /// first. Without `since`, the members that exist. `None` when pruning
    /// removed deletions after `since`, or after `issued` if given.
    pub async fn pim_changes(
        &self,
        collection_id: i64,
        since: Option<i64>,
        issued: Option<i64>,
    ) -> DbResult<Option<Vec<(String, i64, bool)>>> {
        let c = self.conn.lock().await;
        if let Some(seq) = issued.or(since) {
            let pruned: Option<i64> = c
                .query_row(
                    "SELECT pruned_seq FROM pim_collections WHERE id = ?1",
                    [collection_id],
                    |r| r.get(0),
                )
                .optional()?;
            // A collection deleted meanwhile has no valid token either.
            if pruned.is_none_or(|p| seq < p) {
                return Ok(None);
            }
        }
        // Two statements, not one with `?2 IS NULL OR ...`: the OR keeps
        // SQLite from using the seq range of the index.
        let map = |r: &rusqlite::Row| Ok((r.get(0)?, r.get(1)?, r.get(2)?));
        let rows: DbResult<Vec<_>> = match since {
            None => c
                .prepare_cached(
                    "SELECT name, seq, 0 FROM pim_changes
                     WHERE collection_id = ?1 AND deleted = 0 ORDER BY seq",
                )?
                .query_map([collection_id], map)?
                .collect(),
            Some(seq) => c
                .prepare_cached(
                    "SELECT name, seq, deleted != 0 FROM pim_changes
                     WHERE collection_id = ?1 AND seq > ?2 ORDER BY seq",
                )?
                .query_map(params![collection_id, seq], map)?
                .collect(),
        };
        rows.map(Some)
    }

    /// Moves an object to `to_name` in collection `to`, which may be the
    /// same collection. `Updated` means an object at the destination was
    /// replaced.
    pub async fn pim_move_object(
        &self,
        from: i64,
        name: &str,
        to: i64,
        to_name: &str,
        overwrite: bool,
        cond: &Precondition,
    ) -> DbResult<PimWrite> {
        if from == to && name == to_name {
            return Ok(PimWrite::Updated);
        }
        let mut c = self.conn.lock().await;
        let tx = c.transaction()?;
        let source: Option<(String, String)> = tx
            .query_row(
                "SELECT etag, uid FROM pim_objects WHERE collection_id = ?1 AND name = ?2",
                params![from, name],
                |r| Ok((r.get(0)?, r.get(1)?)),
            )
            .optional()?;
        let Some((etag, uid)) = source else {
            return Ok(PimWrite::NotFound);
        };
        if !cond.allows(Some(&etag)) {
            return Ok(PimWrite::PreconditionFailed);
        }
        let taken: bool = tx.query_row(
            "SELECT EXISTS (SELECT 1 FROM pim_objects WHERE collection_id = ?1 AND name = ?2)",
            params![to, to_name],
            |r| r.get(0),
        )?;
        if taken && !overwrite {
            return Ok(PimWrite::PreconditionFailed);
        }
        let holder: Option<String> = tx
            .query_row(
                "SELECT name FROM pim_objects WHERE collection_id = ?1 AND uid = ?2
                     AND name != ?3 AND NOT (collection_id = ?4 AND name = ?5)",
                params![to, uid, to_name, from, name],
                |r| r.get(0),
            )
            .optional()?;
        if let Some(holder) = holder {
            return Ok(PimWrite::UidConflict(holder));
        }
        tx.execute(
            "DELETE FROM pim_objects WHERE collection_id = ?1 AND name = ?2",
            params![to, to_name],
        )?;
        tx.execute(
            "UPDATE pim_objects SET collection_id = ?3, name = ?4, modified_at = ?5
             WHERE collection_id = ?1 AND name = ?2",
            params![from, name, to, to_name, now()],
        )?;
        record_pim_change(&tx, from, name, true)?;
        record_pim_change(&tx, to, to_name, false)?;
        tx.commit()?;
        Ok(if taken {
            PimWrite::Updated
        } else {
            PimWrite::Created
        })
    }

    // ---------- principals, sharing, rooms ----------

    /// An account, room or resource by URL name. Disabled accounts are
    /// invisible.
    pub async fn pim_principal(&self, name: &str) -> DbResult<Option<PimPrincipal>> {
        self.row(
            &format!("SELECT {PRINCIPAL_COLS} FROM {PRINCIPALS} WHERE p.name = ?1 AND {VISIBLE}"),
            [name],
            map_principal,
        )
        .await
    }

    /// Every principal, or with `visible_only` all but disabled accounts.
    pub async fn pim_principals(&self, visible_only: bool) -> DbResult<Vec<PimPrincipal>> {
        let filter = if visible_only { VISIBLE } else { "1" };
        self.rows(
            &format!("SELECT {PRINCIPAL_COLS} FROM {PRINCIPALS} WHERE {filter} ORDER BY p.id"),
            [],
            map_principal,
        )
        .await
    }

    /// The principal of an account.
    pub async fn principal_of(&self, user_id: i64) -> DbResult<i64> {
        self.row(
            "SELECT id FROM principals WHERE user_id = ?1",
            [user_id],
            |r| r.get(0),
        )
        .await?
        .ok_or(rusqlite::Error::QueryReturnedNoRows)
    }

    /// Whether an account, room or resource has this name.
    pub async fn name_taken(&self, name: &str) -> DbResult<bool> {
        Ok(self
            .row("SELECT 1 FROM principals WHERE name = ?1", [name], |_| {
                Ok(())
            })
            .await?
            .is_some())
    }

    /// Every collection of `kind` lent to `user_id`.
    pub async fn pim_shared_collections(
        &self,
        user_id: i64,
        kind: PimKind,
    ) -> DbResult<Vec<(PimCollection, String, PimShareMode)>> {
        self.rows(
            &pim_shared_sql("ORDER BY c.id"),
            (user_id, pim_kind_str(kind)),
            map_shared,
        )
        .await
    }

    /// The collection `id` of `kind` as lent to `user_id`.
    pub async fn pim_shared_collection(
        &self,
        user_id: i64,
        kind: PimKind,
        id: i64,
    ) -> DbResult<Option<(PimCollection, String, PimShareMode)>> {
        self.row(
            &pim_shared_sql("AND c.id = ?3"),
            (user_id, pim_kind_str(kind), id),
            map_shared,
        )
        .await
    }

    /// The owner and kind of a collection.
    pub async fn pim_collection_by_id(
        &self,
        id: i64,
    ) -> DbResult<Option<(i64, PimKind, PimCollection)>> {
        self.row(
            &format!(
                "SELECT {PIM_COLLECTION_COLS}, c.principal_id, c.kind FROM pim_collections c
                 WHERE c.id = ?1"
            ),
            [id],
            |r| Ok((r.get(10)?, wire(r, 11, pim_kind)?, map_pim_collection(r)?)),
        )
        .await
    }

    pub async fn pim_create_link(
        &self,
        collection_id: i64,
        token: &str,
        busy_only: bool,
        expires_at: Option<&str>,
        password_hash: Option<&str>,
    ) -> DbResult<PimLink> {
        let c = self.conn.lock().await;
        c.query_row(
            &format!(
                "INSERT INTO pim_links
                   (token, collection_id, busy_only, created_at, expires_at, password_hash)
                 VALUES (?1, ?2, ?3, ?4, ?5, ?6) RETURNING {PIM_LINK_COLS}"
            ),
            params![
                token,
                collection_id,
                busy_only,
                now(),
                expires_at,
                password_hash
            ],
            map_pim_link,
        )
    }

    /// Per collection of `principal_id`: how many loans and feed links it has.
    pub async fn pim_share_counts(
        &self,
        principal_id: i64,
    ) -> DbResult<std::collections::HashMap<i64, (usize, usize)>> {
        let counts = self
            .rows(
                "SELECT c.id,
                     (SELECT COUNT(*) FROM pim_shares s WHERE s.collection_id = c.id),
                     (SELECT COUNT(*) FROM pim_links l WHERE l.collection_id = c.id)
                 FROM pim_collections c WHERE c.principal_id = ?1",
                [principal_id],
                |r| {
                    Ok((
                        r.get(0)?,
                        (r.get::<_, i64>(1)? as usize, r.get::<_, i64>(2)? as usize),
                    ))
                },
            )
            .await?;
        Ok(counts.into_iter().collect())
    }

    pub async fn pim_links(&self, collection_id: i64) -> DbResult<Vec<PimLink>> {
        self.rows(
            &format!("SELECT {PIM_LINK_COLS} FROM pim_links WHERE collection_id = ?1 ORDER BY id"),
            [collection_id],
            map_pim_link,
        )
        .await
    }

    /// `None` also when the owner's account is disabled.
    pub async fn pim_link_by_token(&self, token: &str) -> DbResult<Option<PimLink>> {
        self.row(
            &format!(
                "SELECT {PIM_LINK_COLS} FROM pim_links WHERE token = ?1 AND collection_id IN (
                     SELECT c.id FROM pim_collections c JOIN principals p ON p.id = c.principal_id
                     LEFT JOIN users u ON u.id = p.user_id WHERE {VISIBLE})"
            ),
            [token],
            map_pim_link,
        )
        .await
    }

    /// Revokes a feed link: one of `collection`, or with `None` whoever made
    /// it. `false` means no row matched.
    pub async fn pim_delete_link(&self, id: i64, collection: Option<i64>) -> DbResult<bool> {
        Ok(self
            .exec(
                "DELETE FROM pim_links WHERE id = ?1 AND (?2 IS NULL OR collection_id = ?2)",
                (id, collection),
            )
            .await?
            > 0)
    }

    /// `(user id, name, mode)` of everyone a collection is lent to.
    pub async fn pim_shares(
        &self,
        collection_id: i64,
    ) -> DbResult<Vec<(i64, String, PimShareMode)>> {
        self.rows(
            "SELECT u.id, u.name, s.mode FROM pim_shares s JOIN users u ON u.id = s.user_id
             WHERE s.collection_id = ?1 ORDER BY u.name",
            [collection_id],
            |r| Ok((r.get(0)?, r.get(1)?, wire(r, 2, PimShareMode::from_wire)?)),
        )
        .await
    }

    /// Active accounts a collection could still be lent to: not `me`, not
    /// already a borrower. `(name, display name)`, by name.
    pub async fn pim_share_candidates(
        &self,
        collection_id: i64,
        me: i64,
    ) -> DbResult<Vec<(String, Option<String>)>> {
        self.rows(
            "SELECT u.name, p.display_name FROM users u JOIN principals p ON p.user_id = u.id
             WHERE u.active = 1 AND u.id != ?2
               AND u.id NOT IN (SELECT user_id FROM pim_shares WHERE collection_id = ?1)
             ORDER BY u.name COLLATE NOCASE",
            (collection_id, me),
            |r| Ok((r.get(0)?, r.get(1)?)),
        )
        .await
    }

    /// Lends a collection, or changes the mode of an existing loan.
    pub async fn pim_set_share(
        &self,
        collection_id: i64,
        user_id: i64,
        mode: PimShareMode,
    ) -> DbResult<()> {
        self.exec(
            "INSERT INTO pim_shares (collection_id, user_id, mode) VALUES (?1, ?2, ?3)
             ON CONFLICT (collection_id, user_id) DO UPDATE SET mode = ?3",
            (collection_id, user_id, mode.as_str()),
        )
        .await?;
        Ok(())
    }

    pub async fn pim_remove_share(&self, collection_id: i64, user_id: i64) -> DbResult<bool> {
        Ok(self
            .exec(
                "DELETE FROM pim_shares WHERE collection_id = ?1 AND user_id = ?2",
                (collection_id, user_id),
            )
            .await?
            > 0)
    }

    pub async fn rooms(&self) -> DbResult<Vec<PimPrincipal>> {
        self.rows(
            &format!(
                "SELECT {PRINCIPAL_COLS} FROM {PRINCIPALS} WHERE p.user_id IS NULL ORDER BY p.name"
            ),
            [],
            map_principal,
        )
        .await
    }

    /// A room or resource with its booking calendar. `None` if the name is
    /// taken by any principal.
    pub async fn create_room(
        &self,
        name: &str,
        display_name: &str,
        kind: UserType,
    ) -> DbResult<Option<PimPrincipal>> {
        let mut c = self.conn.lock().await;
        let tx = c.transaction()?;
        let inserted = tx.execute(
            "INSERT INTO principals (kind, name, display_name)
             SELECT ?1, ?2, ?3 WHERE NOT EXISTS (SELECT 1 FROM principals WHERE name = ?2)",
            params![kind_str(kind), name, display_name],
        )?;
        if inserted == 0 {
            return Ok(None);
        }
        let id = tx.last_insert_rowid();
        tx.execute(
            "INSERT INTO pim_collections (principal_id, kind, slug, displayname, components, created_at)
             VALUES (?1, 'cal', 'default', ?2, 'VEVENT', ?3)",
            params![id, display_name, now()],
        )?;
        tx.commit()?;
        Ok(Some(PimPrincipal {
            id,
            user_id: None,
            name: name.to_string(),
            display_name: Some(display_name.to_string()),
            kind,
            active: true,
        }))
    }

    pub async fn set_room_display_name(&self, id: i64, display_name: &str) -> DbResult<bool> {
        Ok(self
            .exec(
                "UPDATE principals SET display_name = ?2 WHERE id = ?1 AND user_id IS NULL",
                (id, display_name),
            )
            .await?
            > 0)
    }

    /// Deletes a room or resource after `ops`, as [`Self::delete_user`] does.
    pub async fn delete_room(&self, id: i64, ops: &[PimOp]) -> DbResult<bool> {
        let mut c = self.conn.lock().await;
        let tx = c.transaction()?;
        apply_ops(&tx, ops)?;
        if tx.execute(
            "DELETE FROM principals WHERE id = ?1 AND user_id IS NULL",
            [id],
        )? == 0
        {
            return Ok(false);
        }
        tx.commit()?;
        self.forget_defaults();
        Ok(true)
    }

    /// A principal by id, a disabled account's too.
    pub async fn pim_principal_by_id(&self, id: i64) -> DbResult<Option<PimPrincipal>> {
        self.row(
            &format!("SELECT {PRINCIPAL_COLS} FROM {PRINCIPALS} WHERE p.id = ?1"),
            [id],
            map_principal,
        )
        .await
    }

    /// `(collection id, object, data)` of the calendar objects of other
    /// principals whose text holds one of `needles`, compared without ASCII
    /// case.
    pub async fn pim_objects_mentioning(
        &self,
        principal_id: i64,
        needles: &[&str],
    ) -> DbResult<Vec<(i64, PimObject, Vec<u8>)>> {
        let c = self.conn.lock().await;
        // Unfolded first: a fold may split the address.
        let flat = "replace(replace(replace(replace(lower(CAST(o.data AS TEXT)),
             char(13, 10, 32), ''), char(13, 10, 9), ''), char(10, 32), ''), char(10, 9), '')";
        let any: Vec<String> = (0..needles.len())
            .map(|i| format!("instr({flat}, ?{}) > 0", i + 2))
            .collect();
        let mut stmt = c.prepare(&format!(
            "SELECT o.collection_id, {PIM_OBJECT_COLS}, o.data
             FROM pim_objects o JOIN pim_collections c ON c.id = o.collection_id
             WHERE c.principal_id != ?1 AND c.kind = 'cal' AND ({})",
            any.join(" OR ")
        ))?;
        let lower: Vec<String> = needles.iter().map(|n| n.to_ascii_lowercase()).collect();
        let mut params: Vec<&dyn rusqlite::ToSql> = vec![&principal_id];
        params.extend(lower.iter().map(|n| n as &dyn rusqlite::ToSql));
        stmt.query_map(params.as_slice(), |r| {
            Ok((r.get(0)?, map_pim_object_at(r, 1)?, r.get(8)?))
        })?
        .collect()
    }

    /// Public feed links with their collection and owner: those of the
    /// account `owner`, or every one for the admin.
    pub async fn pim_links_with_owner(
        &self,
        owner: Option<i64>,
    ) -> DbResult<Vec<PimLinkWithOwner>> {
        self.rows(
            "SELECT l.id, l.token, l.collection_id, l.busy_only, l.created_at, l.expires_at,
                    l.password_hash, coalesce(c.displayname, c.slug), c.kind, u.id, u.name,
                    u.active != 0
             FROM pim_links l
             JOIN pim_collections c ON c.id = l.collection_id
             JOIN principals p ON p.id = c.principal_id
             JOIN users u ON u.id = p.user_id
             WHERE ?1 IS NULL OR u.id = ?1
             ORDER BY u.name COLLATE NOCASE, l.id DESC",
            [owner],
            |r| {
                Ok(PimLinkWithOwner {
                    link: map_pim_link(r)?,
                    collection_name: r.get(7)?,
                    kind: wire(r, 8, pim_kind)?,
                    owner_id: r.get(9)?,
                    owner_name: r.get(10)?,
                    owner_active: r.get(11)?,
                })
            },
        )
        .await
    }

    /// The loans of the account `owner`'s collections:
    /// `(collection id, name, kind, borrower id, borrower name, mode)`.
    pub async fn pim_lends(
        &self,
        owner: i64,
    ) -> DbResult<Vec<(i64, String, PimKind, i64, String, PimShareMode)>> {
        self.rows(
            "SELECT c.id, coalesce(c.displayname, c.slug), c.kind, u.id, u.name, s.mode
             FROM pim_shares s
             JOIN pim_collections c ON c.id = s.collection_id
             JOIN principals p ON p.id = c.principal_id
             JOIN users u ON u.id = s.user_id
             WHERE p.user_id = ?1
             ORDER BY coalesce(c.displayname, c.slug) COLLATE NOCASE, u.name COLLATE NOCASE",
            [owner],
            |r| {
                Ok((
                    r.get(0)?,
                    r.get(1)?,
                    wire(r, 2, pim_kind)?,
                    r.get(3)?,
                    r.get(4)?,
                    wire(r, 5, PimShareMode::from_wire)?,
                ))
            },
        )
        .await
    }

    /// Whether users may create writable (read-write) shares. Off by default;
    /// the admin setting gates it.
    pub async fn allow_writable_shares(&self) -> DbResult<bool> {
        Ok(self.get_setting("allow_writable_shares").await?.as_deref() == Some("1"))
    }

    pub async fn set_allow_writable_shares(&self, v: bool) -> DbResult<()> {
        self.set_setting("allow_writable_shares", if v { "1" } else { "0" })
            .await
    }
}

/// Write a new password hash and drop every session that was opened with the
/// old one.
///
/// The two belong together: a password is changed because the old one is
/// suspect (an admin resetting a compromised account), and a session that
/// survives the reset leaves whoever holds it signed in. Takes the
/// transaction so the caller can bundle it with its other edits.
fn set_password(tx: &rusqlite::Transaction<'_>, id: i64, pass_hash: &str) -> DbResult<()> {
    tx.execute(
        "UPDATE users SET pass_hash = ?1 WHERE id = ?2",
        params![pass_hash, id],
    )?;
    tx.execute("DELETE FROM sessions WHERE user_id = ?1", [id])?;
    Ok(())
}

/// Commit a credential change, or undo it if the account would be left with
/// no way to sign in.
///
/// The handlers check the same rules first, so the person gets a message that
/// says which rule. This is the guarantee behind those checks: a count read
/// before the write can already be stale when the write lands, and two
/// requests racing could otherwise take away the last credential between them.
/// `false` means the change was rolled back.
fn commit_if_reachable(tx: rusqlite::Transaction<'_>, user_id: i64) -> DbResult<bool> {
    let (has_password, mode) = tx.query_row(
        "SELECT pass_hash != '', auth_mode FROM users WHERE id = ?1",
        [user_id],
        |r| Ok((r.get::<_, bool>(0)?, wire(r, 1, AuthMode::from_wire)?)),
    )?;
    let passkeys: i64 = tx.query_row(
        "SELECT COUNT(*) FROM passkeys WHERE user_id = ?1",
        [user_id],
        |r| r.get(0),
    )?;
    let reachable = match mode {
        AuthMode::Either => has_password || passkeys > 0,
        AuthMode::Both => has_password && passkeys > 0,
    };
    if !reachable {
        tx.rollback()?;
        return Ok(false);
    }
    tx.commit()?;
    Ok(true)
}

/// Column order matched by [`USER_COLS`].
fn map_user(r: &rusqlite::Row) -> DbResult<User> {
    Ok(User {
        id: r.get(0)?,
        name: r.get(1)?,
        is_admin: r.get(2)?,
        active: r.get(3)?,
        single_click: r.get(4)?,
        thumbnails: r.get(5)?,
        language: r.get(6)?,
        default_root_id: r.get(7)?,
        auth_mode: wire(r, 8, AuthMode::from_wire)?,
        has_password: r.get(9)?,
        week_start: r.get(10)?,
    })
}

const PASSKEY_COLS: &str = "id, name, created_at, last_used_at, discoverable, passkey";

fn map_passkey(r: &rusqlite::Row) -> DbResult<PasskeyRow> {
    Ok(PasskeyRow {
        id: r.get(0)?,
        name: r.get(1)?,
        created_at: r.get(2)?,
        last_used_at: r.get(3)?,
        discoverable: r.get(4)?,
        passkey: r.get(5)?,
    })
}

fn insert_roots(
    tx: &rusqlite::Transaction,
    user_id: i64,
    roots: &[(String, Mode)],
) -> DbResult<()> {
    for (path, mode) in roots {
        tx.execute(
            "INSERT INTO user_roots (user_id, path, mode) VALUES (?1, ?2, ?3)",
            params![user_id, path, mode.as_str()],
        )?;
    }
    Ok(())
}

/// `id, path, mode` of `user_roots` starting at column `at`. `None` when the
/// id is NULL: a user without roots in a LEFT JOIN.
fn map_root_at(r: &rusqlite::Row, at: usize) -> DbResult<Option<RootRow>> {
    let Some(id) = r.get(at)? else {
        return Ok(None);
    };
    Ok(Some(RootRow {
        id,
        path: r.get(at + 1)?,
        mode: wire(r, at + 2, Mode::from_wire)?,
    }))
}

const APP_PASSWORD_COLS: &str = "id, name, created_at, last_used_at";

fn map_app_password(r: &rusqlite::Row) -> DbResult<AppPasswordInfo> {
    Ok(AppPasswordInfo {
        id: r.get(0)?,
        name: r.get(1)?,
        created_at: r.get(2)?,
        last_used_at: r.get(3)?,
    })
}

/// Messages an inbox keeps; older ones are dropped.
const INBOX_KEEP: i64 = 100;

fn ensure_inbox(c: &Connection, principal_id: i64) -> DbResult<()> {
    c.prepare_cached(
        "INSERT INTO pim_collections (principal_id, kind, slug, displayname, components, created_at)
         SELECT ?1, 'cal', 'inbox', 'Inbox', 'VEVENT,VTODO,VJOURNAL', ?2
         WHERE NOT EXISTS (SELECT 1 FROM pim_collections
                           WHERE principal_id = ?1 AND kind = 'cal' AND slug = 'inbox')",
    )?
    .execute(params![principal_id, now()])?;
    Ok(())
}

/// The writes of [`PimOp`]s, inside the caller's transaction.
fn apply_ops(tx: &rusqlite::Transaction, ops: &[PimOp]) -> DbResult<()> {
    for op in ops {
        match op {
            PimOp::Put {
                collection_id,
                obj,
                data,
            } => {
                put_object(tx, *collection_id, obj, data)?;
            }
            PimOp::Delete {
                collection_id,
                name,
            } => {
                let gone = tx.execute(
                    "DELETE FROM pim_objects WHERE collection_id = ?1 AND name = ?2",
                    params![collection_id, name],
                )?;
                if gone > 0 {
                    record_pim_change(tx, *collection_id, name, true)?;
                }
            }
            PimOp::Inbox {
                principal_id,
                obj,
                data,
            } => {
                // A room gets its inbox with its first message.
                ensure_inbox(tx, *principal_id)?;
                let inbox: i64 = tx.query_row(
                    "SELECT id FROM pim_collections
                     WHERE principal_id = ?1 AND kind = 'cal' AND slug = 'inbox'",
                    [principal_id],
                    |r| r.get(0),
                )?;
                put_object(tx, inbox, obj, data)?;
                // ponytail: a fixed cap. Clients that never empty the inbox
                // would fill it forever; an age limit may suit better.
                let old: Vec<String> = tx
                    .prepare_cached(
                        "DELETE FROM pim_objects WHERE collection_id = ?1 AND id NOT IN (
                             SELECT id FROM pim_objects WHERE collection_id = ?1
                             ORDER BY id DESC LIMIT ?2)
                         RETURNING name",
                    )?
                    .query_map(params![inbox, INBOX_KEEP], |r| r.get(0))?
                    .collect::<DbResult<_>>()?;
                for name in old {
                    record_pim_change(tx, inbox, &name, true)?;
                }
            }
            PimOp::DeleteCollection(id) => {
                tx.execute("DELETE FROM pim_collections WHERE id = ?1", [id])?;
            }
        }
    }
    Ok(())
}

/// Stores an object under `obj.name`, replacing one of that name.
fn put_object(
    tx: &rusqlite::Transaction,
    collection_id: i64,
    obj: &PimObject,
    data: &[u8],
) -> DbResult<()> {
    tx.execute(
        "INSERT INTO pim_objects (collection_id, name, uid, component, data, etag, modified_at,
             schedule_tag)
         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)
         ON CONFLICT (collection_id, name) DO UPDATE SET uid = ?3, component = ?4,
             data = ?5, etag = ?6, modified_at = ?7, schedule_tag = ?8",
        params![
            collection_id,
            obj.name,
            obj.uid,
            obj.component,
            data,
            obj.etag,
            now(),
            obj.schedule_tag
        ],
    )?;
    record_pim_change(tx, collection_id, &obj.name, false)
}

/// Bumps the collection's `seq` and records it as the latest change of `name`.
fn record_pim_change(
    tx: &rusqlite::Transaction,
    collection_id: i64,
    name: &str,
    deleted: bool,
) -> DbResult<()> {
    let seq: i64 = tx.query_row(
        "UPDATE pim_collections SET seq = seq + 1 WHERE id = ?1 RETURNING seq",
        [collection_id],
        |r| r.get(0),
    )?;
    tx.execute(
        "INSERT INTO pim_changes (collection_id, name, seq, deleted) VALUES (?1, ?2, ?3, ?4)
         ON CONFLICT (collection_id, name) DO UPDATE SET seq = ?3, deleted = ?4",
        params![collection_id, name, seq, deleted],
    )?;
    Ok(())
}

/// An account, room or resource, as CalDAV sees it.
#[derive(Debug, Clone)]
pub struct PimPrincipal {
    pub id: i64,
    /// The account of a person; rooms and resources have none.
    pub user_id: Option<i64>,
    /// The URL segment.
    pub name: String,
    pub display_name: Option<String>,
    pub kind: UserType,
    /// False for a disabled account.
    pub active: bool,
}

impl PimPrincipal {
    pub fn display(&self) -> &str {
        self.display_name.as_deref().unwrap_or(&self.name)
    }
}

// The last column repeats VISIBLE.
const PRINCIPAL_COLS: &str =
    "p.id, p.user_id, p.name, p.display_name, p.kind, (p.user_id IS NULL OR u.active = 1)";
const PRINCIPALS: &str = "principals p LEFT JOIN users u ON u.id = p.user_id";
/// Disabled accounts are hidden.
const VISIBLE: &str = "(p.user_id IS NULL OR u.active = 1)";

fn map_principal(r: &rusqlite::Row) -> DbResult<PimPrincipal> {
    Ok(PimPrincipal {
        id: r.get(0)?,
        user_id: r.get(1)?,
        name: r.get(2)?,
        display_name: r.get(3)?,
        kind: wire(r, 4, |s| match s {
            "person" => Some(UserType::Individual),
            "room" => Some(UserType::Room),
            "resource" => Some(UserType::Resource),
            _ => None,
        })?,
        active: r.get(5)?,
    })
}

fn kind_str(kind: UserType) -> &'static str {
    match kind {
        UserType::Individual => "person",
        UserType::Room => "room",
        UserType::Resource => "resource",
    }
}

/// Read with the table aliased as `c`.
const PIM_COLLECTION_COLS: &str = "c.id, c.slug, c.displayname, c.description, c.color,
     c.timezone, c.sort_order, c.components, c.seq, c.transparent";

fn write_props(
    tx: &rusqlite::Transaction,
    place: PropPlace,
    set: &[DeadProp],
    remove: &[(String, String)],
) -> DbResult<()> {
    let key = place.key();
    let (principal_id, collection_id) = place.owners();
    for (ns, name) in remove {
        tx.execute(
            "DELETE FROM pim_props WHERE place = ?1 AND ns = ?2 AND name = ?3",
            params![key, ns, name],
        )?;
    }
    for p in set {
        tx.execute(
            "INSERT OR REPLACE INTO pim_props (place, principal_id, collection_id, ns, name, xml)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
            params![key, principal_id, collection_id, p.ns, p.name, p.xml],
        )?;
    }
    Ok(())
}

/// Collections lent to `?1` of kind `?2`, with the owner's name and the mode.
fn pim_shared_sql(tail: &str) -> String {
    format!(
        "SELECT {PIM_COLLECTION_COLS}, p.name, s.mode
         FROM pim_shares s
         JOIN pim_collections c ON c.id = s.collection_id
         JOIN principals p ON p.id = c.principal_id
         LEFT JOIN users u ON u.id = p.user_id
         WHERE {VISIBLE} AND s.user_id = ?1 AND c.kind = ?2 {tail}"
    )
}

fn map_shared(r: &rusqlite::Row) -> DbResult<(PimCollection, String, PimShareMode)> {
    Ok((
        map_pim_collection(r)?,
        r.get(10)?,
        wire(r, 11, PimShareMode::from_wire)?,
    ))
}

fn map_pim_collection(r: &rusqlite::Row) -> DbResult<PimCollection> {
    Ok(PimCollection {
        id: r.get(0)?,
        slug: r.get(1)?,
        displayname: r.get(2)?,
        description: r.get(3)?,
        color: r.get(4)?,
        timezone: r.get(5)?,
        sort_order: r.get(6)?,
        components: r.get(7)?,
        seq: r.get(8)?,
        transparent: r.get(9)?,
    })
}

/// Read with the table aliased as `o`.
const PIM_OBJECT_COLS: &str =
    "o.name, o.uid, o.component, o.etag, length(o.data), o.modified_at, o.schedule_tag";

fn map_pim_object(r: &rusqlite::Row) -> DbResult<PimObject> {
    map_pim_object_at(r, 0)
}

/// [`PIM_OBJECT_COLS`] starting at column `at`.
fn map_pim_object_at(r: &rusqlite::Row, at: usize) -> DbResult<PimObject> {
    Ok(PimObject {
        name: r.get(at)?,
        uid: r.get(at + 1)?,
        component: r.get(at + 2)?,
        etag: r.get(at + 3)?,
        size: r.get(at + 4)?,
        modified_at: r.get(at + 5)?,
        schedule_tag: r.get(at + 6)?,
    })
}

const PIM_LINK_COLS: &str =
    "id, token, collection_id, busy_only, created_at, expires_at, password_hash";

fn map_pim_link(r: &rusqlite::Row) -> DbResult<PimLink> {
    Ok(PimLink {
        id: r.get(0)?,
        token: r.get(1)?,
        collection_id: r.get(2)?,
        busy_only: r.get(3)?,
        created_at: r.get(4)?,
        expires_at: r.get(5)?,
        password_hash: r.get(6)?,
    })
}

/// Qualified by the table name, not an alias, so `RETURNING` can use them too.
const SHARE_COLS: &str = "shares.id, shares.token, shares.creator_id, shares.target,
     shares.is_file, shares.mode, shares.created_at, shares.expires_at, shares.password_hash";

fn map_share(r: &rusqlite::Row) -> DbResult<ShareRow> {
    Ok(ShareRow {
        id: r.get(0)?,
        token: r.get(1)?,
        creator_id: r.get(2)?,
        target: r.get(3)?,
        is_file: r.get::<_, i64>(4)? != 0,
        mode: wire(r, 5, Mode::from_wire)?,
        created_at: r.get(6)?,
        expires_at: r.get(7)?,
        password_hash: r.get(8)?,
    })
}

/// A hash of a random string nobody knows. Verified against when the login
/// name does not exist, so both paths cost one Argon2 run.
static DUMMY_HASH: std::sync::LazyLock<String> = std::sync::LazyLock::new(|| {
    crate::auth::hash_password(&crate::auth::random_token()).expect("argon2 hash")
});

/// How long an unlock row outlives its cookie. The cookie dies with the
/// browser, so this only bounds the rows left behind by closed sessions.
const UNLOCK_MAX_AGE_DAYS: i64 = 7;

/// How often [`Db::sweep`] runs. A share the picker can only set to the
/// minute is gone from the owner's list about when it says it is.
const SWEEP_EVERY: std::time::Duration = std::time::Duration::from_secs(60);

/// Run [`Db::sweep`] until the process ends. A failed pass is logged and
/// retried on the next one: nothing downstream depends on it having run.
pub async fn sweep_forever(db: Db) {
    loop {
        tokio::time::sleep(SWEEP_EVERY).await;
        match db.sweep().await {
            Ok((0, 0)) => {}
            Ok((shares, unlocks)) => tracing::debug!(shares, unlocks, "swept expired shares"),
            Err(e) => tracing::warn!(error = %e, "the share sweep failed"),
        }
    }
}

/// The time `ago` in the past as a stored stamp. The format sorts
/// lexicographically, so SQL can compare stamps as text.
fn stamp(ago: chrono::Duration) -> String {
    (chrono::Utc::now() - ago).to_rfc3339_opts(chrono::SecondsFormat::Secs, true)
}

fn now() -> String {
    stamp(chrono::Duration::zero())
}

/// Deletion records kept per collection for sync tokens.
const PIM_TOMBSTONES_KEPT: i64 = 1000;

/// The schema, one step per version. Version `n` has run the first `n`.
/// Append only: a database records how many steps it has run.
const MIGRATIONS: &[&str] = &[
    SCHEMA_V1,
    // A disabled flag so admins can suspend accounts without deleting them.
    "ALTER TABLE users ADD COLUMN active INTEGER NOT NULL DEFAULT 1",
    // Click-to-open mode. Everyone starts on the default (off: single click
    // selects, double click opens).
    "ALTER TABLE users ADD COLUMN single_click INTEGER NOT NULL DEFAULT 0",
    // NULL means "follow the browser".
    "ALTER TABLE users ADD COLUMN language TEXT",
    // The share list is queried by creator on every shares page.
    "CREATE INDEX IF NOT EXISTS idx_shares_creator ON shares(creator_id)",
    // Unlocks cascade with their share, which cascades with its creator's
    // account.
    "ALTER TABLE shares ADD COLUMN password_hash TEXT;
     CREATE TABLE IF NOT EXISTS share_unlocks (
         token TEXT PRIMARY KEY,
         share_id INTEGER NOT NULL REFERENCES shares(id) ON DELETE CASCADE,
         created_at TEXT NOT NULL
     );",
    // `delete_share` cascades into share_unlocks, which is a full scan of
    // that table without this.
    "CREATE INDEX IF NOT EXISTS idx_share_unlocks_share ON share_unlocks(share_id)",
    // On by default, so `--cache` is the only step needed to get thumbnails.
    "ALTER TABLE users ADD COLUMN thumbnails INTEGER NOT NULL DEFAULT 1",
    // No foreign key on purpose: removing a root must not fail because of
    // this column.
    "ALTER TABLE users ADD COLUMN default_root_id INTEGER",
    // Passkeys, and how they combine with the password.
    //
    // `webauthn_id` is the WebAuthn user handle: a random uuid the
    // authenticator stores inside a discoverable credential and hands back
    // at sign-in. It must never change once a passkey exists, or that
    // passkey can no longer be traced to its account. Filled in lazily on
    // the first registration, so accounts that never use a passkey keep it
    // NULL.
    "ALTER TABLE users ADD COLUMN auth_mode TEXT NOT NULL DEFAULT 'either';
     ALTER TABLE users ADD COLUMN webauthn_id TEXT;
     CREATE TABLE IF NOT EXISTS passkeys (
         id INTEGER PRIMARY KEY AUTOINCREMENT,
         user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
         cred_id BLOB NOT NULL UNIQUE,
         passkey TEXT NOT NULL,
         name TEXT NOT NULL,
         discoverable INTEGER,
         created_at TEXT NOT NULL,
         last_used_at TEXT
     );
     CREATE INDEX IF NOT EXISTS idx_passkeys_user ON passkeys(user_id);
     CREATE UNIQUE INDEX IF NOT EXISTS idx_users_webauthn_id
         ON users(webauthn_id) WHERE webauthn_id IS NOT NULL;",
    // `secret_hash` is UNIQUE because the lookup keys on it.
    "CREATE TABLE IF NOT EXISTS app_passwords (
         id INTEGER PRIMARY KEY AUTOINCREMENT,
         user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
         name TEXT NOT NULL,
         secret_hash TEXT NOT NULL UNIQUE,
         created_at TEXT NOT NULL,
         last_used_at TEXT
     );
     CREATE INDEX IF NOT EXISTS idx_app_passwords_user ON app_passwords(user_id);",
    // CalDAV and CardDAV. A principal owns collections: every account has
    // one, and rooms and resources are principals without an account, so
    // they share the account name space but no account query can return
    // them. The trigger gives each new account its principal and refuses a
    // name a room already has.
    //
    // `seq` counts every change to a collection and its members;
    // `pim_changes` keeps the latest change per member, deletions included,
    // for sync tokens. `pim_shares` lends a collection to another account.
    // `schedule_tag` is NULL for objects that schedule nothing. A
    // `transparent` calendar adds no busy time to scheduling (RFC 6638
    // `schedule-calendar-transp`). `pim_links` are public feeds; not rows of
    // `shares`, because every path-based share query would then have to
    // skip them. `pim_props` holds the properties clients set that the
    // server does not interpret, as XML, per collection, home or principal.
    "CREATE TABLE IF NOT EXISTS principals (
         id INTEGER PRIMARY KEY AUTOINCREMENT,
         user_id INTEGER UNIQUE REFERENCES users(id) ON DELETE CASCADE,
         kind TEXT NOT NULL CHECK (kind IN ('person','room','resource')),
         name TEXT NOT NULL UNIQUE COLLATE NOCASE,
         display_name TEXT,
         CHECK ((kind = 'person') = (user_id IS NOT NULL))
     );
     INSERT INTO principals (user_id, kind, name)
         SELECT id, 'person', name FROM users;
     CREATE TRIGGER IF NOT EXISTS principal_of_user AFTER INSERT ON users
     BEGIN
         INSERT INTO principals (user_id, kind, name)
             VALUES (NEW.id, 'person', NEW.name);
     END;
     CREATE TABLE IF NOT EXISTS pim_collections (
         id INTEGER PRIMARY KEY AUTOINCREMENT,
         principal_id INTEGER NOT NULL
             REFERENCES principals(id) ON DELETE CASCADE,
         kind TEXT NOT NULL CHECK (kind IN ('cal','card')),
         slug TEXT NOT NULL,
         displayname TEXT,
         description TEXT,
         color TEXT,
         timezone TEXT,
         sort_order TEXT,
         components TEXT NOT NULL DEFAULT '',
         transparent INTEGER NOT NULL DEFAULT 0,
         seq INTEGER NOT NULL DEFAULT 0,
         created_at TEXT NOT NULL,
         UNIQUE (principal_id, kind, slug)
     );
     CREATE TABLE IF NOT EXISTS pim_objects (
         id INTEGER PRIMARY KEY AUTOINCREMENT,
         collection_id INTEGER NOT NULL
             REFERENCES pim_collections(id) ON DELETE CASCADE,
         name TEXT NOT NULL,
         uid TEXT NOT NULL,
         component TEXT NOT NULL,
         data BLOB NOT NULL,
         etag TEXT NOT NULL,
         modified_at TEXT NOT NULL,
         schedule_tag TEXT,
         UNIQUE (collection_id, name),
         UNIQUE (collection_id, uid)
     );
     CREATE TABLE IF NOT EXISTS pim_changes (
         collection_id INTEGER NOT NULL
             REFERENCES pim_collections(id) ON DELETE CASCADE,
         name TEXT NOT NULL,
         seq INTEGER NOT NULL,
         deleted INTEGER NOT NULL,
         PRIMARY KEY (collection_id, name)
     );
     CREATE INDEX IF NOT EXISTS idx_pim_changes_seq
         ON pim_changes(collection_id, seq);
     CREATE TABLE IF NOT EXISTS pim_shares (
         collection_id INTEGER NOT NULL
             REFERENCES pim_collections(id) ON DELETE CASCADE,
         user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
         mode TEXT NOT NULL CHECK (mode IN ('ro','rw','rw+schedule')),
         PRIMARY KEY (collection_id, user_id)
     );
     CREATE INDEX IF NOT EXISTS idx_pim_shares_user ON pim_shares(user_id);
     CREATE TABLE IF NOT EXISTS pim_links (
         id INTEGER PRIMARY KEY AUTOINCREMENT,
         token TEXT NOT NULL UNIQUE,
         collection_id INTEGER NOT NULL
             REFERENCES pim_collections(id) ON DELETE CASCADE,
         busy_only INTEGER NOT NULL DEFAULT 0,
         created_at TEXT NOT NULL,
         expires_at TEXT,
         password_hash TEXT
     );
     CREATE INDEX IF NOT EXISTS idx_pim_links_collection
         ON pim_links(collection_id);
     CREATE TABLE IF NOT EXISTS pim_props (
         place TEXT NOT NULL,
         principal_id INTEGER REFERENCES principals(id) ON DELETE CASCADE,
         collection_id INTEGER REFERENCES pim_collections(id) ON DELETE CASCADE,
         ns TEXT NOT NULL,
         name TEXT NOT NULL,
         xml TEXT NOT NULL,
         PRIMARY KEY (place, ns, name),
         CHECK ((principal_id IS NULL) != (collection_id IS NULL))
     );",
    // Monday, the ISO 8601 week start.
    "ALTER TABLE users ADD COLUMN week_start INTEGER NOT NULL DEFAULT 1",
    // The calendar that receives invitations (RFC 6638,
    // schedule-default-calendar-URL). NULL: the oldest one.
    "ALTER TABLE principals ADD COLUMN default_calendar_id INTEGER
         REFERENCES pim_collections(id) ON DELETE SET NULL",
    // Sync tokens below this are refused: the tombstones they need are gone.
    "ALTER TABLE pim_collections ADD COLUMN pruned_seq INTEGER NOT NULL DEFAULT 0",
    // The cascades from a deleted principal or collection.
    "CREATE INDEX IF NOT EXISTS idx_pim_props_principal ON pim_props(principal_id);
     CREATE INDEX IF NOT EXISTS idx_pim_props_collection ON pim_props(collection_id);",
];

const SCHEMA_V1: &str = r#"
CREATE TABLE IF NOT EXISTS users (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    name TEXT NOT NULL UNIQUE COLLATE NOCASE,
    pass_hash TEXT NOT NULL,
    is_admin INTEGER NOT NULL DEFAULT 0,
    created_at TEXT NOT NULL
);

CREATE TABLE IF NOT EXISTS user_roots (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
    path TEXT NOT NULL,
    mode TEXT NOT NULL CHECK (mode IN ('rw','ro')),
    UNIQUE (user_id, path)
);

CREATE TABLE IF NOT EXISTS shares (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    token TEXT NOT NULL UNIQUE,
    creator_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
    target TEXT NOT NULL,
    is_file INTEGER NOT NULL,
    mode TEXT NOT NULL CHECK (mode IN ('rw','ro')),
    created_at TEXT NOT NULL,
    expires_at TEXT
);

CREATE TABLE IF NOT EXISTS sessions (
    token TEXT PRIMARY KEY,
    user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
    created_at TEXT NOT NULL,
    last_seen_at TEXT NOT NULL
);

CREATE TABLE IF NOT EXISTS settings (
    key TEXT PRIMARY KEY,
    value TEXT NOT NULL
);
INSERT OR IGNORE INTO settings (key, value) VALUES ('allow_writable_shares', '0');
"#;

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

    // Most tests use an in-memory DB (the file-based path is still covered
    // by `v1_db_migrates_to_v2` and the integration harness' `Db::open`).
    async fn mem() -> Db {
        Db::open(Path::new(":memory:")).await.unwrap()
    }

    /// `update_user` is the only way production edits these fields, so the
    /// tests exercise that path rather than per-field helpers.
    async fn edit(db: &Db, id: i64, pass: Option<&str>, admin: Option<bool>, active: Option<bool>) {
        db.update_user(id, pass, admin, active, None).await.unwrap();
    }

    /// Backdate a stamp, which production code has no reason to do.
    async fn set_last_used(db: &Db, secret_hash: &str, at: &str) {
        db.conn
            .lock()
            .await
            .execute(
                "UPDATE app_passwords SET last_used_at = ?1 WHERE secret_hash = ?2",
                params![at, secret_hash],
            )
            .unwrap();
    }

    async fn db_with_admin() -> (Db, User) {
        let db = mem().await;
        let hash = crate::auth::hash_password("admin1234").unwrap();
        let admin = db.create_admin("admin", &hash).await.unwrap().unwrap();
        (db, admin)
    }

    #[tokio::test]
    async fn pim_patch_keeps_concurrent_changes() {
        let (db, admin) = db_with_admin().await;
        let pid = db.principal_of(admin.id).await.unwrap();
        db.pim_ensure_defaults(pid).await.unwrap();
        let before = db.pim_collections(pid, PimKind::Calendar).await.unwrap()[0].clone();
        let place = PropPlace::Collection(before.id);
        let named = PimCollection {
            displayname: Some("Work".into()),
            ..before.clone()
        };
        let colored = PimCollection {
            color: Some("#ff0000".into()),
            ..before.clone()
        };
        // Two writers that both read `before`.
        db.pim_patch(place, Some((&before, &named)), &[], &[])
            .await
            .unwrap();
        db.pim_patch(place, Some((&before, &colored)), &[], &[])
            .await
            .unwrap();
        let (_, _, now) = db.pim_collection_by_id(before.id).await.unwrap().unwrap();
        assert_eq!(now.displayname.as_deref(), Some("Work"));
        assert_eq!(now.color.as_deref(), Some("#ff0000"));
        assert_eq!(now.seq, before.seq + 2);
    }

    #[tokio::test]
    async fn fresh_db_state() {
        let db = mem().await;
        assert_eq!(db.user_count().await.unwrap(), 0);
        assert_eq!(db.count_admins().await.unwrap(), 0);
        assert!(!db.allow_writable_shares().await.unwrap());
        assert!(db.find_user_by_name("nobody").await.unwrap().is_none());
        assert!(db.find_user_by_id(1).await.unwrap().is_none());
        assert!(db.all_users_with_roots().await.unwrap().is_empty());
    }

    #[tokio::test]
    async fn app_passwords_open_one_account_and_stamp_their_use() {
        let (db, admin) = db_with_admin().await;
        let secret = crate::auth::short_token();
        let hash = crate::auth::app_password_hash(&secret);
        let row = db
            .add_app_password(admin.id, "laptop", &hash)
            .await
            .unwrap()
            .unwrap();
        assert_eq!(row.name, "laptop");
        assert!(row.last_used_at.is_none());

        let hit = db.user_by_app_password(&hash).await.unwrap().unwrap();
        assert_eq!(hit.id, admin.id);
        let listed = |db: &Db| {
            let db = db.clone();
            async move {
                db.app_passwords(admin.id).await.unwrap()[0]
                    .last_used_at
                    .clone()
            }
        };
        assert!(listed(&db).await.is_some(), "first use must stamp the row");

        // Ten minutes back is inside the hour window, and far enough from
        // `now()` that a rewrite would show at second resolution.
        let inside = stamp(chrono::Duration::minutes(10));
        set_last_used(&db, &hash, &inside).await;
        db.user_by_app_password(&hash).await.unwrap().unwrap();
        assert_eq!(
            listed(&db).await.as_deref(),
            Some(inside.as_str()),
            "a second use inside the hour wrote the stamp again"
        );

        let outside = stamp(chrono::Duration::hours(2));
        set_last_used(&db, &hash, &outside).await;
        db.user_by_app_password(&hash).await.unwrap().unwrap();
        assert_ne!(
            listed(&db).await.as_deref(),
            Some(outside.as_str()),
            "a use after the window left the stamp stale"
        );

        // The raw secret is not the key, and an inactive account does not match.
        let other = crate::auth::app_password_hash(&crate::auth::short_token());
        assert!(db.user_by_app_password(&other).await.unwrap().is_none());
        assert!(db.user_by_app_password(&secret).await.unwrap().is_none());
        edit(&db, admin.id, None, None, Some(false)).await;
        assert!(db.user_by_app_password(&hash).await.unwrap().is_none());
        edit(&db, admin.id, None, None, Some(true)).await;
        assert!(db.user_by_app_password(&hash).await.unwrap().is_some());

        for i in 1..APP_PASSWORD_LIMIT {
            let h = crate::auth::app_password_hash(&crate::auth::short_token());
            assert!(
                db.add_app_password(admin.id, &format!("c{i}"), &h)
                    .await
                    .unwrap()
                    .is_some()
            );
        }
        let h = crate::auth::app_password_hash(&crate::auth::short_token());
        assert!(
            db.add_app_password(admin.id, "one-too-many", &h)
                .await
                .unwrap()
                .is_none()
        );

        assert!(db.delete_app_password(row.id, admin.id).await.unwrap());
        assert!(!db.delete_app_password(row.id, admin.id).await.unwrap());
        assert!(db.user_by_app_password(&hash).await.unwrap().is_none());
    }

    #[tokio::test]
    async fn an_admin_password_reset_revokes_the_app_passwords() {
        let (db, admin) = db_with_admin().await;
        let hash = crate::auth::app_password_hash(&crate::auth::short_token());
        db.add_app_password(admin.id, "mount", &hash)
            .await
            .unwrap()
            .unwrap();

        // An edit that sets no password leaves them alone.
        edit(&db, admin.id, None, None, Some(true)).await;
        assert_eq!(db.app_passwords(admin.id).await.unwrap().len(), 1);

        // A reset is meant to revoke every way in that existed before it.
        let fresh = crate::auth::hash_password("rescued12").unwrap();
        edit(&db, admin.id, Some(&fresh), None, None).await;
        assert!(db.app_passwords(admin.id).await.unwrap().is_empty());
        assert!(db.user_by_app_password(&hash).await.unwrap().is_none());
    }

    #[tokio::test]
    async fn v1_db_migrates_to_v2() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("legacy.sqlite");
        {
            let conn = rusqlite::Connection::open(&path).unwrap();
            conn.execute_batch(SCHEMA_V1).unwrap();
            conn.execute(
                "INSERT INTO users (name, pass_hash, is_admin, created_at)
                 VALUES ('legacy', 'hash', 1, '2024-01-01T00:00:00Z')",
                [],
            )
            .unwrap();
            conn.execute(
                "INSERT INTO user_roots (user_id, path, mode) VALUES (1, 'docs', 'rw')",
                [],
            )
            .unwrap();
        }
        let db = Db::open(&path).await.unwrap();
        assert_eq!(db.user_count().await.unwrap(), 1);
        let u = db.find_user_by_name("legacy").await.unwrap().unwrap();
        assert!(u.active, "v2 migration must default active to true");
        assert!(u.is_admin);
        assert_eq!(db.user_roots(u.id).await.unwrap().len(), 1);
        // Accounts from before CalDAV get their principal.
        let p = db.pim_principal("legacy").await.unwrap().unwrap();
        assert_eq!(p.user_id, Some(u.id));
        assert_eq!(db.principal_of(u.id).await.unwrap(), p.id);
        let indexes: i64 = db
            .conn
            .lock()
            .await
            .query_row(
                "SELECT COUNT(*) FROM sqlite_master WHERE type = 'index'
                 AND name IN ('idx_pim_props_principal', 'idx_pim_props_collection')",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(indexes, 2);
        // Migrations are idempotent.
        let db2 = Db::open(&path).await.unwrap();
        assert_eq!(db2.user_count().await.unwrap(), 1);
        assert!(
            db2.find_user_by_name("legacy")
                .await
                .unwrap()
                .unwrap()
                .active
        );
    }

    #[tokio::test]
    async fn deleted_defaults_come_back() {
        let (db, admin) = db_with_admin().await;
        let p = db.principal_of(admin.id).await.unwrap();
        db.pim_ensure_defaults(p).await.unwrap();
        let books = db.pim_collections(p, PimKind::Addressbook).await.unwrap();
        assert_eq!(books.len(), 1);
        db.pim_apply(&[PimOp::DeleteCollection(books[0].id)])
            .await
            .unwrap();
        // Remembered defaults must not hide that one is gone.
        db.pim_ensure_defaults(p).await.unwrap();
        assert_eq!(
            db.pim_collections(p, PimKind::Addressbook)
                .await
                .unwrap()
                .len(),
            1
        );
    }

    #[tokio::test]
    async fn rooms_are_principals_not_accounts() {
        let (db, admin) = db_with_admin().await;
        let bob = db.create_user("bob", "hash", false, &[]).await.unwrap();
        let room = db
            .create_room("board", "Board", UserType::Room)
            .await
            .unwrap()
            .unwrap();
        assert_eq!(room.user_id, None);

        // No account query returns the room.
        assert_eq!(db.user_count().await.unwrap(), 2);
        assert!(db.find_user_by_name("board").await.unwrap().is_none());
        assert!(db.verify_password("board", "").await.unwrap().is_none());
        let names: Vec<_> = db.all_users_with_roots().await.unwrap();
        assert!(names.iter().all(|(u, _)| u.name != "board"));

        // One name space, without case.
        assert!(db.name_taken("BOARD").await.unwrap());
        assert!(db.create_user("Board", "hash", false, &[]).await.is_err());
        assert!(
            db.create_room("BOB", "Bob", UserType::Resource)
                .await
                .unwrap()
                .is_none()
        );

        // Deleting an account takes its principal, collections and loans.
        let bob_p = db.principal_of(bob.id).await.unwrap();
        db.pim_ensure_defaults(bob_p).await.unwrap();
        let admin_p = db.principal_of(admin.id).await.unwrap();
        db.pim_ensure_defaults(admin_p).await.unwrap();
        let bobs = db.pim_collections(bob_p, PimKind::Calendar).await.unwrap();
        let admins = db
            .pim_collections(admin_p, PimKind::Calendar)
            .await
            .unwrap();
        db.pim_set_share(admins[0].id, bob.id, PimShareMode::Ro)
            .await
            .unwrap();
        assert!(db.delete_user(bob.id, &[]).await.unwrap());
        assert!(db.pim_principal("bob").await.unwrap().is_none());
        assert!(!db.name_taken("bob").await.unwrap());
        assert!(db.pim_collection_by_id(bobs[0].id).await.unwrap().is_none());
        assert!(db.pim_shares(admins[0].id).await.unwrap().is_empty());
        // The room outlives it.
        assert!(db.pim_principal("board").await.unwrap().is_some());
    }

    #[tokio::test]
    async fn admin_user_and_passwords() {
        let (db, admin) = db_with_admin().await;
        assert!(admin.is_admin);
        assert!(admin.active);
        // Root "." rw is assigned by create_admin.
        let roots = db.user_roots(admin.id).await.unwrap();
        assert_eq!(roots.len(), 1);
        assert_eq!(roots[0].path, ".");
        assert_eq!(roots[0].mode, Mode::Rw);

        assert!(
            db.verify_password("admin", "admin1234")
                .await
                .unwrap()
                .is_some()
        );
        assert!(db.verify_password("admin", "nope").await.unwrap().is_none());
        // Name lookup is case-insensitive (COLLATE NOCASE).
        assert!(
            db.verify_password("ADMIN", "admin1234")
                .await
                .unwrap()
                .is_some()
        );
        // Disabled users cannot verify.
        edit(&db, admin.id, None, None, Some(false)).await;
        assert!(
            db.verify_password("admin", "admin1234")
                .await
                .unwrap()
                .is_none()
        );
        edit(&db, admin.id, None, None, Some(true)).await;
        assert!(
            db.verify_password("admin", "admin1234")
                .await
                .unwrap()
                .is_some()
        );
    }

    #[tokio::test]
    async fn setup_is_won_by_exactly_one_caller() {
        let db = mem().await;
        let hash = crate::auth::hash_password("admin1234").unwrap();
        assert!(db.create_admin("first", &hash).await.unwrap().is_some());
        // The guard lives in the insert, so a different name loses too.
        assert!(db.create_admin("second", &hash).await.unwrap().is_none());
        assert_eq!(db.user_count().await.unwrap(), 1);
        // The loser rolled back cleanly: no orphaned root row.
        let first = db.find_user_by_name("first").await.unwrap().unwrap();
        assert_eq!(db.user_roots(first.id).await.unwrap().len(), 1);
    }

    #[tokio::test]
    async fn changing_a_password_drops_that_users_sessions() {
        let (db, admin) = db_with_admin().await;
        let h = crate::auth::hash_password("bobpass1").unwrap();
        let bob = db.create_user("bob", &h, false, &[]).await.unwrap();
        db.create_session(admin.id, "admin-tok").await.unwrap();
        db.create_session(bob.id, "bob-tok-1").await.unwrap();
        db.create_session(bob.id, "bob-tok-2").await.unwrap();

        let new_h = crate::auth::hash_password("bobpass2").unwrap();
        edit(&db, bob.id, Some(&new_h), None, None).await;
        assert!(
            db.session_user_with_roots("bob-tok-1")
                .await
                .unwrap()
                .is_none()
        );
        assert!(
            db.session_user_with_roots("bob-tok-2")
                .await
                .unwrap()
                .is_none()
        );
        // Only the reset user is signed out.
        assert!(
            db.session_user_with_roots("admin-tok")
                .await
                .unwrap()
                .is_some()
        );

        // The admin-edit path bundles the same rule into its transaction.
        db.create_session(bob.id, "bob-tok-3").await.unwrap();
        let h3 = crate::auth::hash_password("bobpass3").unwrap();
        db.update_user(bob.id, Some(&h3), None, None, None)
            .await
            .unwrap();
        assert!(
            db.session_user_with_roots("bob-tok-3")
                .await
                .unwrap()
                .is_none()
        );
        // An edit that leaves the password alone keeps the session.
        db.create_session(bob.id, "bob-tok-4").await.unwrap();
        db.update_user(bob.id, None, Some(true), None, None)
            .await
            .unwrap();
        assert!(
            db.session_user_with_roots("bob-tok-4")
                .await
                .unwrap()
                .is_some()
        );
    }

    #[tokio::test]
    async fn sessions_lifecycle() {
        let (db, admin) = db_with_admin().await;
        assert!(
            db.session_user_with_roots("ghost-token")
                .await
                .unwrap()
                .is_none()
        );
        db.create_session(admin.id, "tok1").await.unwrap();
        let (u, _) = db.session_user_with_roots("tok1").await.unwrap().unwrap();
        assert_eq!(u.id, admin.id);
        // Disabling the user invalidates existing sessions.
        edit(&db, admin.id, None, None, Some(false)).await;
        assert!(db.session_user_with_roots("tok1").await.unwrap().is_none());
        edit(&db, admin.id, None, None, Some(true)).await;
        assert!(db.session_user_with_roots("tok1").await.unwrap().is_some());
        db.delete_session("tok1").await.unwrap();
        assert!(db.session_user_with_roots("tok1").await.unwrap().is_none());
    }

    #[tokio::test]
    async fn user_crud_and_roots() {
        let (db, _admin) = db_with_admin().await;
        let h = crate::auth::hash_password("bobpass1").unwrap();
        let bob = db
            .create_user("bob", &h, false, &[("docs".into(), Mode::Rw)])
            .await
            .unwrap();
        assert!(!bob.is_admin);
        assert!(bob.active);

        // Duplicate name (case-insensitive) is rejected.
        let h2 = crate::auth::hash_password("carolpass1").unwrap();
        assert!(db.create_user("BOB", &h2, false, &[]).await.is_err());
        assert!(db.create_user("carol", &h2, false, &[]).await.is_ok());

        // Lookup helpers.
        assert_eq!(
            db.find_user_by_name("Bob").await.unwrap().unwrap().id,
            bob.id
        );
        assert_eq!(
            db.find_user_by_id(bob.id).await.unwrap().unwrap().name,
            "bob"
        );
        assert!(db.find_user_by_name("dave").await.unwrap().is_none());
        assert_eq!(db.all_users_with_roots().await.unwrap().len(), 3);

        // Root replacement semantics.
        let roots = db.user_roots(bob.id).await.unwrap();
        assert_eq!(roots.len(), 1);
        db.update_user(
            bob.id,
            None,
            None,
            None,
            Some(&[(".".into(), Mode::Ro), ("docs".into(), Mode::Rw)]),
        )
        .await
        .unwrap();
        let roots = db.user_roots(bob.id).await.unwrap();
        assert_eq!(roots.len(), 2);
        assert!(roots.iter().any(|r| r.path == "." && r.mode == Mode::Ro));
        db.update_user(bob.id, None, None, None, Some(&[]))
            .await
            .unwrap();
        assert!(db.user_roots(bob.id).await.unwrap().is_empty());

        // Password update.
        let new_h = crate::auth::hash_password("bobpass2").unwrap();
        edit(&db, bob.id, Some(&new_h), None, None).await;
        assert!(
            db.verify_password("bob", "bobpass1")
                .await
                .unwrap()
                .is_none()
        );
        assert!(
            db.verify_password("bob", "bobpass2")
                .await
                .unwrap()
                .is_some()
        );

        // Admin flag + count (only active admins count).
        edit(&db, bob.id, None, Some(true), None).await;
        assert_eq!(db.count_admins().await.unwrap(), 2);
        edit(&db, bob.id, None, None, Some(false)).await;
        assert_eq!(db.count_admins().await.unwrap(), 1);
        edit(&db, bob.id, None, Some(false), None).await;

        // Deletion.
        assert!(db.delete_user(bob.id, &[]).await.unwrap());
        assert!(db.find_user_by_id(bob.id).await.unwrap().is_none());
        assert!(!db.delete_user(bob.id, &[]).await.unwrap());
        assert_eq!(db.user_count().await.unwrap(), 2);
    }

    fn share_row(expires_at: Option<&str>) -> ShareRow {
        ShareRow {
            id: 1,
            token: "t".into(),
            creator_id: 1,
            target: "docs".into(),
            is_file: false,
            mode: Mode::Ro,
            created_at: "2024-01-01T00:00:00Z".into(),
            expires_at: expires_at.map(str::to_string),
            password_hash: None,
        }
    }

    #[test]
    fn share_expiry_logic() {
        assert!(!share_row(None).is_expired());
        assert!(!share_row(Some("2999-01-01T00:00:00Z")).is_expired());
        assert!(share_row(Some("2000-01-01T00:00:00Z")).is_expired());
        // Unparseable expiry → treated as not expired (fail open for reads).
        assert!(!share_row(Some("not-a-date")).is_expired());
    }

    #[tokio::test]
    async fn shares_crud() {
        let (db, admin) = db_with_admin().await;
        let s1 = db
            .create_share(admin.id, "tok-a", "docs", false, Mode::Ro, None, None)
            .await
            .unwrap();
        let s2 = db
            .create_share(
                admin.id,
                "tok-b",
                "file.txt",
                true,
                Mode::Rw,
                Some("2999-01-01T00:00:00Z"),
                None,
            )
            .await
            .unwrap();
        assert!(s2.id > s1.id);

        let found = db.share_by_token("tok-b").await.unwrap().unwrap();
        assert!(found.is_file);
        assert_eq!(found.mode, Mode::Rw);
        assert!(db.share_by_token("nope").await.unwrap().is_none());

        // Listed newest-first.
        let list = db.user_shares(admin.id).await.unwrap();
        assert_eq!(list.len(), 2);
        assert_eq!(list[0].id, s2.id);
        // Other users see nothing.
        let h = crate::auth::hash_password("bobpass1").unwrap();
        let bob = db.create_user("bob", &h, false, &[]).await.unwrap();
        assert!(db.user_shares(bob.id).await.unwrap().is_empty());

        // Only the creator can delete.
        assert!(!db.delete_share(s1.id, Some(bob.id)).await.unwrap());
        assert!(db.delete_share(s1.id, Some(admin.id)).await.unwrap());
        assert!(db.share_by_token("tok-a").await.unwrap().is_none());
        assert!(!db.delete_share(s1.id, Some(admin.id)).await.unwrap());
    }

    /// The sweep decides which timestamps are past, and `expires_at` is
    /// stored in whatever RFC 3339 shape the client sent.
    #[tokio::test]
    async fn the_sweep_drops_expired_shares_and_stale_unlocks() {
        let (db, admin) = db_with_admin().await;
        let make = async |token: &str, expires: Option<&str>| {
            db.create_share(admin.id, token, "docs", false, Mode::Ro, expires, None)
                .await
                .unwrap()
        };
        let past_offset = make("gone-offset", Some("2000-01-01T00:00:00+02:00")).await;
        make("gone-utc", Some("2000-01-01T00:00:00Z")).await;
        let future = make("stays-future", Some("2999-01-01T00:00:00Z")).await;
        make("stays-forever", None).await;
        // `is_expired` keeps an unreadable timestamp; the sweep must agree,
        // or the two disagree about the same row.
        make("stays-garbage", Some("not-a-date")).await;

        let fresh = db.create_share_unlock(future.id).await.unwrap();
        let stale = db.create_share_unlock(future.id).await.unwrap();
        let doomed = db.create_share_unlock(past_offset.id).await.unwrap();
        {
            let c = db.conn.lock().await;
            c.execute(
                "UPDATE share_unlocks SET created_at = '2000-01-01T00:00:00Z' WHERE token = ?1",
                [&stale],
            )
            .unwrap();
        }

        let (shares, unlocks) = db.sweep().await.unwrap();
        assert_eq!(shares, 2, "only the two past timestamps");
        // The stale one, plus the cascade from the expired share it opened.
        assert_eq!(unlocks, 1, "the cascade runs with the share, not here");

        for token in ["gone-offset", "gone-utc"] {
            assert!(db.share_by_token(token).await.unwrap().is_none(), "{token}");
        }
        for token in ["stays-future", "stays-forever", "stays-garbage"] {
            assert!(db.share_by_token(token).await.unwrap().is_some(), "{token}");
        }

        assert!(db.share_unlock_valid(&fresh, future.id).await.unwrap());
        assert!(!db.share_unlock_valid(&stale, future.id).await.unwrap());
        assert!(
            !db.share_unlock_valid(&doomed, past_offset.id)
                .await
                .unwrap(),
            "an unlock must not outlive the share it opened"
        );

        assert_eq!(
            db.sweep().await.unwrap(),
            (0, 0),
            "a second pass is a no-op"
        );
    }

    /// The unlock token is what a visitor's cookie carries, so an unlock
    /// that opened the wrong share would be a full bypass of the password.
    #[tokio::test]
    async fn share_unlocks_are_bound_to_one_share() {
        let (db, admin) = db_with_admin().await;
        let a = db
            .create_share(
                admin.id,
                "tok-a",
                "docs",
                false,
                Mode::Ro,
                None,
                Some("hash"),
            )
            .await
            .unwrap();
        let b = db
            .create_share(
                admin.id,
                "tok-b",
                "other",
                false,
                Mode::Ro,
                None,
                Some("hash"),
            )
            .await
            .unwrap();
        assert_eq!(
            db.share_by_token("tok-a")
                .await
                .unwrap()
                .unwrap()
                .password_hash,
            Some("hash".to_string())
        );

        let unlock = db.create_share_unlock(a.id).await.unwrap();
        assert!(db.share_unlock_valid(&unlock, a.id).await.unwrap());
        assert!(!db.share_unlock_valid(&unlock, b.id).await.unwrap());
        assert!(!db.share_unlock_valid("nonsense", a.id).await.unwrap());

        // Deleting the share takes its unlocks with it, so a re-created
        // share that happened to reuse the id could not inherit them.
        assert!(db.delete_share(a.id, Some(admin.id)).await.unwrap());
        assert!(!db.share_unlock_valid(&unlock, a.id).await.unwrap());
    }

    #[tokio::test]
    async fn revoking_a_path_takes_its_descendants_only() {
        let (db, admin) = db_with_admin().await;
        let mk = async |token: &str, target: &str| {
            db.create_share(admin.id, token, target, false, Mode::Ro, None, None)
                .await
                .unwrap();
        };
        mk("t-self", "docs").await;
        mk("t-child", "docs/a.txt").await;
        mk("t-deep", "docs/inner/b.txt").await;
        // A sibling whose name merely starts with "docs" must survive.
        mk("t-sibling", "docs2/c.txt").await;
        mk("t-other", "src").await;
        // SQL wildcards in a path are literal characters, not patterns.
        mk("t-wild", "do%s/d.txt").await;

        assert_eq!(db.revoke_shares_at("docs").await.unwrap(), 3);
        for gone in ["t-self", "t-child", "t-deep"] {
            assert!(db.share_by_token(gone).await.unwrap().is_none(), "{gone}");
        }
        for kept in ["t-sibling", "t-other", "t-wild"] {
            assert!(db.share_by_token(kept).await.unwrap().is_some(), "{kept}");
        }
        // Revoking a path nobody shared is a no-op, not an error.
        assert_eq!(db.revoke_shares_at("nothing/here").await.unwrap(), 0);
    }

    #[tokio::test]
    async fn settings_round_trip() {
        let (db, _admin) = db_with_admin().await;
        assert!(!db.allow_writable_shares().await.unwrap());
        db.set_allow_writable_shares(true).await.unwrap();
        assert!(db.allow_writable_shares().await.unwrap());
        // Upsert semantics.
        db.set_allow_writable_shares(false).await.unwrap();
        assert!(!db.allow_writable_shares().await.unwrap());
        // Generic get/set.
        db.set_setting("custom", "v").await.unwrap();
        assert_eq!(
            db.get_setting("custom").await.unwrap().as_deref(),
            Some("v")
        );
        assert_eq!(db.get_setting("missing").await.unwrap(), None);
    }
}