soma_auth/

sqlite.rs

use std::fmt::Write as _;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};

use rusqlite::types::Value;
use rusqlite::{Connection, OptionalExtension, params};
use sha2::{Digest, Sha256};
use tracing::warn;

use crate::at_rest::{TokenEncryptionKey, maybe_decrypt_bound, maybe_encrypt_bound};
use crate::error::AuthError;
use crate::types::{
    AllowedUserRow, AuthorizationCodeRow, AuthorizationRequestRow, BrowserLoginStateRow,
    BrowserSessionRow, NativeAuthorizationResultRow, RefreshTokenRow, RegisteredClient,
    UpstreamOauthCredentialRow, UpstreamOauthStateRow,
};

#[path = "sqlite_rows.rs"]
mod sqlite_rows;
use sqlite_rows::{
    row_to_allowed_user, row_to_authorization_code, row_to_authorization_request,
    row_to_browser_login_state, row_to_browser_session, row_to_native_authorization_result,
    row_to_upstream_oauth_credentials, row_to_upstream_oauth_state,
};

const UPSTREAM_OAUTH_STATE_MAX_TTL_SECS: i64 = 600;
/// Schema version for the `PRAGMA user_version` migration guard.
/// Increment this whenever a migration step is added to `run_migrations`.
const SCHEMA_VERSION: i64 = 2;

use crate::util::{
    ensure_restrictive_permissions, fingerprint, now_unix, set_restrictive_permissions,
};

const SQLITE_BUSY_TIMEOUT_MS: u64 = 5_000;
const SQLITE_POOL_SIZE: usize = 4;

#[derive(Clone)]
pub struct SqliteStore {
    conns: Arc<Vec<Mutex<Connection>>>,
    next_conn: Arc<AtomicUsize>,
    path: Arc<PathBuf>,
    /// Optional at-rest encryption key for upstream provider refresh tokens.
    enc_key: Option<Arc<TokenEncryptionKey>>,
}

impl std::fmt::Debug for SqliteStore {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("SqliteStore")
            .field("path", &self.path)
            .field("enc_key", &self.enc_key.as_ref().map(|_| "<redacted>"))
            .finish_non_exhaustive()
    }
}

impl SqliteStore {
    pub async fn open(path: PathBuf) -> Result<Self, AuthError> {
        Self::open_with_key(path, None).await
    }

    pub async fn open_with_key(
        path: PathBuf,
        enc_key: Option<TokenEncryptionKey>,
    ) -> Result<Self, AuthError> {
        let path_for_open = path.clone();
        let conns = tokio::task::spawn_blocking(move || {
            open_connections(path_for_open.as_path(), SQLITE_POOL_SIZE)
        })
        .await;
        let store = match conns {
            Ok(result) => result,
            Err(error) => Err(AuthError::Storage(format!(
                "sqlite open task failed: {error}"
            ))),
        }
        .map(|conns| Self {
            conns: Arc::new(conns.into_iter().map(Mutex::new).collect()),
            next_conn: Arc::new(AtomicUsize::new(0)),
            path: Arc::new(path),
            enc_key: enc_key.map(Arc::new),
        })?;

        store.cleanup_expired().await?;
        Ok(store)
    }

    pub async fn pragma(&self, name: &str) -> Result<String, AuthError> {
        let pragma = match name {
            "journal_mode" | "busy_timeout" | "foreign_keys" => name.to_string(),
            other => {
                return Err(AuthError::Config(format!(
                    "unsupported pragma query `{other}`"
                )));
            }
        };

        self.with_conn(move |conn| {
            conn.query_row(&format!("PRAGMA {pragma};"), [], |row| {
                row.get::<_, Value>(0)
            })
            .map(|value| match value {
                Value::Text(text) => text,
                Value::Integer(int) => int.to_string(),
                other => format!("{other:?}"),
            })
            .map_err(sqlite_error)
        })
        .await
    }

    pub async fn register_client(&self, client: RegisteredClient) -> Result<(), AuthError> {
        self.with_conn(move |conn| {
            let redirect_uris = serde_json::to_string(&client.redirect_uris)
                .map_err(|error| AuthError::Storage(format!("serialize redirect_uris: {error}")))?;
            conn.execute(
                "INSERT INTO registered_clients (client_id, redirect_uris, created_at)
                 VALUES (?1, ?2, ?3)
                 ON CONFLICT(client_id) DO UPDATE SET
                    redirect_uris = excluded.redirect_uris,
                    created_at = excluded.created_at",
                params![client.client_id, redirect_uris, client.created_at],
            )
            .map_err(sqlite_error)?;
            Ok(())
        })
        .await
    }

    pub async fn find_client(
        &self,
        client_id: &str,
    ) -> Result<Option<RegisteredClient>, AuthError> {
        let client_id = client_id.to_string();
        self.with_conn(move |conn| {
            conn.query_row(
                "SELECT client_id, redirect_uris, created_at
                 FROM registered_clients
                 WHERE client_id = ?1",
                params![client_id],
                |row| {
                    let redirect_uris: String = row.get(1)?;
                    let redirect_uris = serde_json::from_str(&redirect_uris).map_err(|error| {
                        rusqlite::Error::FromSqlConversionFailure(
                            1,
                            rusqlite::types::Type::Text,
                            Box::new(error),
                        )
                    })?;
                    Ok(RegisteredClient {
                        client_id: row.get(0)?,
                        redirect_uris,
                        created_at: row.get(2)?,
                    })
                },
            )
            .optional()
            .map_err(sqlite_error)
        })
        .await
    }

    pub async fn insert_authorization_request(
        &self,
        request: AuthorizationRequestRow,
    ) -> Result<(), AuthError> {
        self.with_conn(move |conn| {
            conn.execute(
                "INSERT INTO authorization_requests (
                    state, client_id, redirect_uri, client_state, resource, scope, provider_code_verifier,
                    code_challenge, code_challenge_method, created_at, expires_at, provider
                 ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12)",
                params![
                    request.state,
                    request.client_id,
                    request.redirect_uri,
                    request.client_state,
                    request.resource,
                    request.scope,
                    request.provider_code_verifier,
                    request.code_challenge,
                    request.code_challenge_method,
                    request.created_at,
                    request.expires_at,
                    request.provider,
                ],
            )
            .map_err(sqlite_error)?;
            Ok(())
        })
        .await
    }

    pub async fn take_authorization_request(
        &self,
        state: &str,
    ) -> Result<AuthorizationRequestRow, AuthError> {
        let state = state.to_string();
        let now = now_unix();
        self.with_conn(move |conn| {
            conn.query_row(
                "DELETE FROM authorization_requests
                 WHERE state = ?1
                   AND expires_at > ?2
                 RETURNING state, client_id, redirect_uri, client_state, scope, provider_code_verifier,
                           code_challenge, code_challenge_method, created_at, expires_at, resource, provider",
                params![state, now],
                row_to_authorization_request,
            )
            .map_err(|error| match error {
                rusqlite::Error::QueryReturnedNoRows => AuthError::InvalidGrant(
                    "authorization state is missing, expired, or already used".to_string(),
                ),
                other => sqlite_error(other),
            })
        })
        .await
    }

    pub async fn insert_auth_code(&self, code: AuthorizationCodeRow) -> Result<(), AuthError> {
        self.with_conn(move |conn| {
            conn.execute(
                "INSERT INTO authorization_codes (
                    code, client_id, subject, redirect_uri, resource, scope,
                    code_challenge, code_challenge_method, provider_refresh_token,
                    created_at, expires_at, provider
                 ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12)",
                params![
                    code.code,
                    code.client_id,
                    code.subject,
                    code.redirect_uri,
                    code.resource,
                    code.scope,
                    code.code_challenge,
                    code.code_challenge_method,
                    code.provider_refresh_token,
                    code.created_at,
                    code.expires_at,
                    code.provider,
                ],
            )
            .map_err(sqlite_error)?;
            Ok(())
        })
        .await
    }

    pub async fn redeem_auth_code(&self, code: &str) -> Result<AuthorizationCodeRow, AuthError> {
        let code = code.to_string();
        let now = now_unix();
        self.with_conn(move |conn| {
            conn.query_row(
                "DELETE FROM authorization_codes
                 WHERE code = ?1
                   AND expires_at > ?2
                 RETURNING code, client_id, subject, redirect_uri, scope,
                           code_challenge, code_challenge_method, provider_refresh_token,
                           created_at, expires_at, resource, provider",
                params![code, now],
                row_to_authorization_code,
            )
            .map_err(|error| match error {
                rusqlite::Error::QueryReturnedNoRows => AuthError::InvalidGrant(
                    "authorization code is missing, expired, or already redeemed".to_string(),
                ),
                other => sqlite_error(other),
            })
        })
        .await
    }

    /// Insert a new refresh token row, storing a SHA-256 hash of the raw token
    /// as the primary key.  The plaintext token is **never** persisted; only the
    /// caller-returned value contains it.  If an encryption key is configured,
    /// `provider_refresh_token` is encrypted at rest before storage.
    ///
    /// Use [`Self::rotate_refresh_token`] instead of calling this twice when replacing
    /// an existing token — that method performs the swap atomically.
    pub async fn upsert_refresh_token(&self, token: RefreshTokenRow) -> Result<(), AuthError> {
        let hash = hash_token(&token.refresh_token);
        let encrypted_provider_rt = token
            .provider_refresh_token
            .as_deref()
            .map(|raw| maybe_encrypt_bound(self.enc_key.as_deref(), raw, &refresh_token_aad(&hash)))
            .transpose()?;
        self.with_conn(move |conn| {
            conn.execute(
                "INSERT INTO refresh_tokens (
                    refresh_token_hash, client_id, subject, resource, scope,
                    provider_refresh_token, created_at, expires_at, provider
                 ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)
                 ON CONFLICT(refresh_token_hash) DO UPDATE SET
                    client_id = excluded.client_id,
                    subject = excluded.subject,
                    resource = excluded.resource,
                    scope = excluded.scope,
                    provider_refresh_token = excluded.provider_refresh_token,
                    created_at = excluded.created_at,
                    expires_at = excluded.expires_at,
                    provider = excluded.provider",
                params![
                    hash,
                    token.client_id,
                    token.subject,
                    token.resource,
                    token.scope,
                    encrypted_provider_rt,
                    token.created_at,
                    token.expires_at,
                    token.provider,
                ],
            )
            .map_err(sqlite_error)?;
            Ok(())
        })
        .await
    }

    /// Atomically replace an existing refresh token with a new one in a single
    /// SQLite transaction.  The old token is deleted and the new token is
    /// inserted; if the old token is not found or has expired the operation
    /// fails without inserting the new row (replay-safe).
    ///
    /// Both the DELETE and the INSERT are wrapped in an explicit `BEGIN` /
    /// `COMMIT` so a crash between the two statements cannot leave the database
    /// without a valid refresh token.
    ///
    /// Returns the newly issued `RefreshTokenRow` (with `refresh_token` set to
    /// the new plaintext value) on success.
    pub async fn rotate_refresh_token(
        &self,
        old_token: &str,
        new_token: RefreshTokenRow,
    ) -> Result<Option<RefreshTokenRow>, AuthError> {
        let old_hash = hash_token(old_token);
        let new_hash = hash_token(&new_token.refresh_token);
        let now = now_unix();
        let encrypted_provider_rt = new_token
            .provider_refresh_token
            .as_deref()
            .map(|raw| {
                maybe_encrypt_bound(self.enc_key.as_deref(), raw, &refresh_token_aad(&new_hash))
            })
            .transpose()?;
        self.with_conn(move |conn| {
            conn.execute_batch("BEGIN").map_err(sqlite_error)?;

            let delete_result = conn
                .execute(
                    "DELETE FROM refresh_tokens
                     WHERE refresh_token_hash = ?1
                       AND expires_at > ?2",
                    params![old_hash, now],
                )
                .map_err(sqlite_error);

            let deleted = match delete_result {
                Ok(n) => n,
                Err(e) => {
                    drop(conn.execute_batch("ROLLBACK"));
                    return Err(e);
                }
            };

            if deleted == 0 {
                // Old token not found or already expired — rollback and reject.
                drop(conn.execute_batch("ROLLBACK"));
                return Ok(None);
            }

            let insert_result = conn
                .execute(
                    "INSERT INTO refresh_tokens (
                    refresh_token_hash, client_id, subject, resource, scope,
                    provider_refresh_token, created_at, expires_at, provider
                 ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)",
                    params![
                        new_hash,
                        new_token.client_id,
                        new_token.subject,
                        new_token.resource,
                        new_token.scope,
                        encrypted_provider_rt,
                        new_token.created_at,
                        new_token.expires_at,
                        new_token.provider,
                    ],
                )
                .map_err(sqlite_error);

            match insert_result {
                Ok(_) => {
                    conn.execute_batch("COMMIT").map_err(sqlite_error)?;
                    Ok(Some(new_token))
                }
                Err(e) => {
                    drop(conn.execute_batch("ROLLBACK"));
                    Err(e)
                }
            }
        })
        .await
    }

    pub async fn find_refresh_token(
        &self,
        refresh_token: &str,
    ) -> Result<Option<RefreshTokenRow>, AuthError> {
        let hash = hash_token(refresh_token);
        // Keep the plaintext value in memory so the caller receives a row with
        // `refresh_token` populated (the DB never stores it).
        let plaintext = refresh_token.to_string();
        let now = now_unix();
        let enc_key = self.enc_key.clone();
        self.with_conn(move |conn| {
            let row = conn
                .query_row(
                    "SELECT client_id, subject, scope,
                        provider_refresh_token, created_at, expires_at, resource, provider
                 FROM refresh_tokens
                 WHERE refresh_token_hash = ?1
                   AND expires_at > ?2",
                    params![hash, now],
                    |row| {
                        Ok(RefreshTokenRow {
                            refresh_token: plaintext.clone(),
                            client_id: row.get(0)?,
                            subject: row.get(1)?,
                            scope: row.get(2)?,
                            provider_refresh_token: row.get(3)?,
                            created_at: row.get(4)?,
                            expires_at: row.get(5)?,
                            resource: row.get(6).unwrap_or_default(),
                            provider: row.get(7)?,
                        })
                    },
                )
                .optional()
                .map_err(sqlite_error)?;

            // Decrypt provider_refresh_token if present and an enc key is
            // configured.  maybe_decrypt_bound is a no-op for plaintext
            // values, so this is safe to call unconditionally once a row is
            // found.  The AAD re-derives the row identity from the same hash
            // used for the lookup, so a ciphertext transplanted onto a
            // different row fails authentication here.
            match row {
                Some(mut r) => {
                    if let Some(raw) = r.provider_refresh_token.as_deref() {
                        r.provider_refresh_token = Some(maybe_decrypt_bound(
                            enc_key.as_deref(),
                            raw,
                            &refresh_token_aad(&hash),
                        )?);
                    }
                    Ok(Some(r))
                }
                None => Ok(None),
            }
        })
        .await
    }

    /// Whether any unexpired refresh token has ever been issued, for
    /// any client. This is a single-tenant, admin-only gateway, so "someone
    /// already completed the Google consent screen once" is a reasonable
    /// proxy for "we don't need to force full re-consent again" without
    /// having to know which subject is about to authenticate.
    ///
    /// No longer called internally — `authorize()` now uses the
    /// provider-scoped [`Self::has_any_refresh_token_for_provider`] instead
    /// (this unscoped version incorrectly treats "some OTHER provider
    /// already has a refresh token on file" as a reason to skip forced
    /// consent on a user's very first login with a *different* provider).
    /// Retained as general-purpose public API for other consumers of this
    /// shared crate, not because removing the internal call site was an
    /// accident — do not assume this is dead code to delete.
    pub async fn has_any_refresh_token(&self) -> Result<bool, AuthError> {
        let now = now_unix();
        self.with_conn(move |conn| {
            conn.query_row(
                "SELECT EXISTS(SELECT 1 FROM refresh_tokens WHERE expires_at > ?1)",
                params![now],
                |row| row.get::<_, i64>(0),
            )
            .map(|count| count != 0)
            .map_err(sqlite_error)
        })
        .await
    }

    /// Same as [`Self::has_any_refresh_token`], scoped to one provider.
    ///
    /// `authorize()` (Task 11) uses this instead of the unscoped version to
    /// decide whether to force the upstream consent screen — the unscoped
    /// version incorrectly treats "Google already has a refresh token on
    /// file" as a reason to skip forced consent on a user's very first
    /// Authelia or GitHub login, silently degrading that new provider's
    /// first session to no local refresh token.
    pub async fn has_any_refresh_token_for_provider(
        &self,
        provider: &str,
    ) -> Result<bool, AuthError> {
        let provider = provider.to_string();
        let now = now_unix();
        self.with_conn(move |conn| {
            conn.query_row(
                "SELECT EXISTS(SELECT 1 FROM refresh_tokens WHERE provider = ?1 AND expires_at > ?2)",
                params![provider, now],
                |row| row.get::<_, i64>(0),
            )
            .map(|count| count != 0)
            .map_err(sqlite_error)
        })
        .await
    }

    pub async fn upsert_browser_session(
        &self,
        session: BrowserSessionRow,
    ) -> Result<(), AuthError> {
        self.with_conn(move |conn| {
            conn.execute(
                "INSERT INTO browser_sessions (
                    session_id, subject, email, csrf_token, created_at, expires_at
                 ) VALUES (?1, ?2, ?3, ?4, ?5, ?6)
                 ON CONFLICT(session_id) DO UPDATE SET
                    subject = excluded.subject,
                    email = excluded.email,
                    csrf_token = excluded.csrf_token,
                    created_at = excluded.created_at,
                    expires_at = excluded.expires_at",
                params![
                    session.session_id,
                    session.subject,
                    session.email,
                    session.csrf_token,
                    session.created_at,
                    session.expires_at,
                ],
            )
            .map_err(sqlite_error)?;
            Ok(())
        })
        .await
    }

    pub async fn find_browser_session(
        &self,
        session_id: &str,
    ) -> Result<Option<BrowserSessionRow>, AuthError> {
        let session_id = session_id.to_string();
        let now = now_unix();
        self.with_conn(move |conn| {
            conn.query_row(
                "SELECT session_id, subject, email, csrf_token, created_at, expires_at
                 FROM browser_sessions
                 WHERE session_id = ?1
                   AND expires_at > ?2",
                params![session_id, now],
                row_to_browser_session,
            )
            .optional()
            .map_err(sqlite_error)
        })
        .await
    }

    pub async fn revoke_browser_session(&self, session_id: &str) -> Result<(), AuthError> {
        let session_id = session_id.to_string();
        self.with_conn(move |conn| {
            conn.execute(
                "DELETE FROM browser_sessions WHERE session_id = ?1",
                params![session_id],
            )
            .map_err(sqlite_error)?;
            Ok(())
        })
        .await
    }

    /// Run an arbitrary SQL batch against the store — test fixtures only.
    ///
    /// Gated behind `cfg(any(test, debug_assertions))` (deliberately not a
    /// Cargo feature, mirroring `upstream::cache`'s test seam) so an
    /// arbitrary-SQL execution method can never ship in
    /// `--all-features --release` artifacts.
    #[cfg(any(test, debug_assertions))]
    pub async fn execute_test_statement(&self, sql: &str) -> Result<(), AuthError> {
        let sql = sql.to_string();
        self.with_conn(move |conn| conn.execute_batch(&sql).map_err(sqlite_error))
            .await
    }

    pub async fn insert_browser_login_state(
        &self,
        login: BrowserLoginStateRow,
    ) -> Result<(), AuthError> {
        self.with_conn(move |conn| {
            conn.execute(
                "INSERT INTO browser_login_states (
                    state, return_to, provider_code_verifier, created_at, expires_at, provider
                 ) VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
                params![
                    login.state,
                    login.return_to,
                    login.provider_code_verifier,
                    login.created_at,
                    login.expires_at,
                    login.provider,
                ],
            )
            .map_err(sqlite_error)?;
            Ok(())
        })
        .await
    }

    pub async fn count_pending_oauth_states(&self) -> Result<usize, AuthError> {
        let now = now_unix();
        self.with_conn(move |conn| {
            let authorization_requests: i64 = conn
                .query_row(
                    "SELECT COUNT(*) FROM authorization_requests WHERE expires_at > ?1",
                    params![now],
                    |row| row.get(0),
                )
                .map_err(sqlite_error)?;
            let browser_login_states: i64 = conn
                .query_row(
                    "SELECT COUNT(*) FROM browser_login_states WHERE expires_at > ?1",
                    params![now],
                    |row| row.get(0),
                )
                .map_err(sqlite_error)?;
            let native_authorization_results: i64 = conn
                .query_row(
                    "SELECT COUNT(*) FROM native_authorization_results WHERE expires_at > ?1",
                    params![now],
                    |row| row.get(0),
                )
                .map_err(sqlite_error)?;
            Ok(
                (authorization_requests + browser_login_states + native_authorization_results)
                    as usize,
            )
        })
        .await
    }

    pub async fn take_browser_login_state(
        &self,
        state: &str,
    ) -> Result<Option<BrowserLoginStateRow>, AuthError> {
        let state = state.to_string();
        let now = now_unix();
        self.with_conn(move |conn| {
            conn.query_row(
                "DELETE FROM browser_login_states
                 WHERE state = ?1
                   AND expires_at > ?2
                 RETURNING state, return_to, provider_code_verifier, created_at, expires_at, provider",
                params![state, now],
                row_to_browser_login_state,
            )
            .optional()
            .map_err(sqlite_error)
        })
        .await
    }

    /// Store a native-flow authorization code keyed by `state`, for the
    /// polling desktop client to retrieve via `take_native_authorization_result`.
    ///
    /// Last-write-wins on a `state` collision (e.g. a client retrying
    /// `/authorize` with the same `state` after a timeout): each row is
    /// single-use (deleted on first successful poll), so overwriting with the
    /// newest code is correct — silently dropping the newest code instead
    /// (`DO NOTHING`) would leave the polling client hung until the row's TTL
    /// expires, with no error surfaced anywhere.
    pub async fn insert_native_authorization_result(
        &self,
        result: NativeAuthorizationResultRow,
    ) -> Result<(), AuthError> {
        self.with_conn(move |conn| {
            conn.execute(
                "INSERT INTO native_authorization_results (state, code, created_at, expires_at)
                 VALUES (?1, ?2, ?3, ?4)
                 ON CONFLICT(state) DO UPDATE SET
                    code = excluded.code,
                    created_at = excluded.created_at,
                    expires_at = excluded.expires_at",
                params![
                    result.state,
                    result.code,
                    result.created_at,
                    result.expires_at,
                ],
            )
            .map_err(sqlite_error)?;
            Ok(())
        })
        .await
    }

    /// One-shot read-and-delete of a pending native-flow authorization code.
    pub async fn take_native_authorization_result(
        &self,
        state: &str,
    ) -> Result<Option<NativeAuthorizationResultRow>, AuthError> {
        let state = state.to_string();
        let now = now_unix();
        self.with_conn(move |conn| {
            conn.query_row(
                "DELETE FROM native_authorization_results
                 WHERE state = ?1
                   AND expires_at > ?2
                 RETURNING state, code, created_at, expires_at",
                params![state, now],
                row_to_native_authorization_result,
            )
            .optional()
            .map_err(sqlite_error)
        })
        .await
    }

    /// Delete expired rows from all short-lived tables. Also drops upstream OAuth
    /// credential rows whose access token has expired AND have no refresh token
    /// available for re-use (SEC-9). Returns the total number of deleted rows.
    pub async fn cleanup_expired(&self) -> Result<u64, AuthError> {
        let now = now_unix();
        self.with_conn(move |conn| {
            let mut total: u64 = 0;
            for table in [
                "authorization_requests",
                "authorization_codes",
                "refresh_tokens",
                "browser_sessions",
                "browser_login_states",
                "native_authorization_results",
            ] {
                let deleted = conn
                    .execute(
                        &format!("DELETE FROM {table} WHERE expires_at <= ?1"),
                        params![now],
                    )
                    .map_err(sqlite_error)?;
                total += deleted as u64;
            }
            let deleted = conn
                .execute(
                    "DELETE FROM upstream_oauth_state WHERE expires_at <= ?1",
                    params![now],
                )
                .map_err(sqlite_error)?;
            total += deleted as u64;
            let deleted = conn
                .execute(
                    "DELETE FROM upstream_oauth_credentials
                     WHERE access_token_expires_at <= ?1 AND refresh_token_present = 0",
                    params![now],
                )
                .map_err(sqlite_error)?;
            total += deleted as u64;
            Ok(total)
        })
        .await
    }

    pub async fn upsert_upstream_oauth_credentials(
        &self,
        row: UpstreamOauthCredentialRow,
    ) -> Result<(), AuthError> {
        self.with_conn(move |conn| {
            conn.execute(
                "INSERT INTO upstream_oauth_credentials (
                    upstream_name, subject, client_id, granted_scopes_json,
                    token_blob, token_blob_nonce, token_received_at,
                    access_token_expires_at, refresh_token_present
                 ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)
                 ON CONFLICT(upstream_name, subject) DO UPDATE SET
                    client_id = excluded.client_id,
                    granted_scopes_json = excluded.granted_scopes_json,
                    token_blob = excluded.token_blob,
                    token_blob_nonce = excluded.token_blob_nonce,
                    token_received_at = excluded.token_received_at,
                    access_token_expires_at = excluded.access_token_expires_at,
                    refresh_token_present = excluded.refresh_token_present",
                params![
                    row.upstream_name,
                    row.subject,
                    row.client_id,
                    row.granted_scopes_json,
                    row.token_blob,
                    row.token_blob_nonce,
                    row.token_received_at,
                    row.access_token_expires_at,
                    i64::from(row.refresh_token_present),
                ],
            )
            .map_err(sqlite_error)?;
            Ok(())
        })
        .await
    }

    pub async fn find_upstream_oauth_credentials(
        &self,
        upstream_name: &str,
        subject: &str,
    ) -> Result<Option<UpstreamOauthCredentialRow>, AuthError> {
        let upstream_name = upstream_name.to_string();
        let subject = subject.to_string();
        self.with_conn(move |conn| {
            conn.query_row(
                "SELECT upstream_name, subject, client_id, granted_scopes_json,
                        token_blob, token_blob_nonce, token_received_at,
                        access_token_expires_at, refresh_token_present
                 FROM upstream_oauth_credentials
                 WHERE upstream_name = ?1 AND subject = ?2",
                params![upstream_name, subject],
                row_to_upstream_oauth_credentials,
            )
            .optional()
            .map_err(sqlite_error)
        })
        .await
    }

    pub async fn delete_upstream_oauth_credentials(
        &self,
        upstream_name: &str,
        subject: &str,
    ) -> Result<(), AuthError> {
        let upstream_name = upstream_name.to_string();
        let subject = subject.to_string();
        self.with_conn(move |conn| {
            conn.execute(
                "DELETE FROM upstream_oauth_credentials
                 WHERE upstream_name = ?1 AND subject = ?2",
                params![upstream_name, subject],
            )
            .map_err(sqlite_error)?;
            Ok(())
        })
        .await
    }

    pub async fn save_upstream_oauth_state(
        &self,
        row: UpstreamOauthStateRow,
    ) -> Result<(), AuthError> {
        if row.expires_at <= row.created_at
            || row.expires_at - row.created_at > UPSTREAM_OAUTH_STATE_MAX_TTL_SECS
        {
            return Err(AuthError::InvalidGrant(
                "state TTL exceeds 600s".to_string(),
            ));
        }
        self.with_conn(move |conn| {
            conn.execute(
                "INSERT INTO upstream_oauth_state (
                    upstream_name, subject, csrf_token, pkce_verifier, created_at, expires_at
                 ) VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
                params![
                    row.upstream_name,
                    row.subject,
                    row.csrf_token,
                    row.pkce_verifier,
                    row.created_at,
                    row.expires_at,
                ],
            )
            .map_err(sqlite_error)?;
            Ok(())
        })
        .await
    }

    pub async fn find_upstream_oauth_state_subject(
        &self,
        upstream_name: &str,
        csrf_token: &str,
        now: i64,
    ) -> Result<Option<String>, AuthError> {
        let upstream_name = upstream_name.to_string();
        let csrf_token = csrf_token.to_string();
        self.with_conn(move |conn| {
            conn.query_row(
                "SELECT subject
                 FROM upstream_oauth_state
                 WHERE upstream_name = ?1
                   AND csrf_token = ?2
                   AND expires_at > ?3",
                params![upstream_name, csrf_token, now],
                |row| row.get(0),
            )
            .optional()
            .map_err(sqlite_error)
        })
        .await
    }

    /// Look up `(upstream_name, subject)` by `csrf_token` alone.
    ///
    /// Used by the OAuth callback handler to recover the upstream identity from
    /// the state parameter without requiring the caller to know it upfront.
    pub async fn find_upstream_oauth_state_owner(
        &self,
        csrf_token: &str,
        now: i64,
    ) -> Result<Option<(String, String)>, AuthError> {
        let csrf_token = csrf_token.to_string();
        self.with_conn(move |conn| {
            conn.query_row(
                "SELECT upstream_name, subject
                 FROM upstream_oauth_state
                 WHERE csrf_token = ?1
                   AND expires_at > ?2",
                params![csrf_token, now],
                |row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)),
            )
            .optional()
            .map_err(sqlite_error)
        })
        .await
    }

    /// Delete a pending OAuth state token by CSRF token to foreclose replay attacks after exchange failure.
    pub async fn delete_upstream_oauth_state_by_csrf(
        &self,
        csrf_token: &str,
        now: i64,
    ) -> Result<(), AuthError> {
        let csrf_token = csrf_token.to_string();
        self.with_conn(move |conn| {
            conn.execute(
                "DELETE FROM upstream_oauth_state
                 WHERE csrf_token = ?1
                   AND expires_at > ?2",
                params![csrf_token, now],
            )
            .map_err(sqlite_error)?;
            Ok(())
        })
        .await
    }

    /// Bind a dynamic OAuth `client_id` to a pending CSRF state row.
    ///
    /// Called by `begin_authorization` after generating the authorization URL
    /// so that `complete_authorization_callback` can later look up which
    /// `client_id` was used for this specific flow (lab-77y5.15).
    pub async fn set_upstream_oauth_state_client_id(
        &self,
        upstream_name: &str,
        csrf_token: &str,
        client_id: &str,
    ) -> Result<(), AuthError> {
        let upstream_name = upstream_name.to_string();
        let csrf_token = csrf_token.to_string();
        let client_id = client_id.to_string();
        self.with_conn(move |conn| {
            conn.execute(
                "UPDATE upstream_oauth_state
                 SET dynamic_client_id = ?1
                 WHERE upstream_name = ?2
                   AND csrf_token = ?3",
                params![client_id, upstream_name, csrf_token],
            )
            .map_err(sqlite_error)?;
            Ok(())
        })
        .await
    }

    /// Retrieve the `dynamic_client_id` bound to a pending CSRF state row.
    ///
    /// Returns `None` when no row matches or the row has expired. Used by
    /// `complete_authorization_callback` to recover the exact `client_id` that
    /// was used when the authorization URL was generated (lab-77y5.15).
    pub async fn get_upstream_oauth_state_client_id(
        &self,
        upstream_name: &str,
        csrf_token: &str,
        now: i64,
    ) -> Result<Option<String>, AuthError> {
        let upstream_name = upstream_name.to_string();
        let csrf_token = csrf_token.to_string();
        self.with_conn(move |conn| {
            conn.query_row(
                "SELECT dynamic_client_id
                 FROM upstream_oauth_state
                 WHERE upstream_name = ?1
                   AND csrf_token = ?2
                   AND expires_at > ?3",
                params![upstream_name, csrf_token, now],
                |row| row.get::<_, Option<String>>(0),
            )
            .optional()
            .map(|opt| opt.flatten())
            .map_err(sqlite_error)
        })
        .await
    }

    /// Atomic take-once via `DELETE ... RETURNING`.
    pub async fn take_upstream_oauth_state(
        &self,
        upstream_name: &str,
        subject: &str,
        csrf_token: &str,
        now: i64,
    ) -> Result<Option<UpstreamOauthStateRow>, AuthError> {
        let upstream_name = upstream_name.to_string();
        let subject = subject.to_string();
        let csrf_token = csrf_token.to_string();
        self.with_conn(move |conn| {
            conn.query_row(
                "DELETE FROM upstream_oauth_state
                 WHERE upstream_name = ?1
                   AND subject = ?2
                   AND csrf_token = ?3
                   AND expires_at > ?4
                 RETURNING upstream_name, subject, csrf_token, pkce_verifier, created_at, expires_at",
                params![upstream_name, subject, csrf_token, now],
                row_to_upstream_oauth_state,
            )
            .optional()
            .map_err(sqlite_error)
        })
        .await
    }

    pub async fn save_dynamic_client_registration(
        &self,
        upstream_name: &str,
        subject: &str,
        client_id: &str,
    ) -> Result<(), AuthError> {
        let upstream_name = upstream_name.to_string();
        let subject = subject.to_string();
        let client_id = client_id.to_string();
        let now = now_unix();
        self.with_conn(move |conn| {
            conn.execute(
                "INSERT INTO upstream_oauth_dynamic_clients (upstream_name, subject, client_id, created_at)
                 VALUES (?1, ?2, ?3, ?4)
                 ON CONFLICT(upstream_name, subject) DO UPDATE SET
                    client_id = excluded.client_id,
                    created_at = excluded.created_at",
                params![upstream_name, subject, client_id, now],
            )
            .map_err(sqlite_error)?;
            Ok(())
        })
        .await
    }

    pub async fn find_dynamic_client_registration(
        &self,
        upstream_name: &str,
        subject: &str,
    ) -> Result<Option<String>, AuthError> {
        let upstream_name = upstream_name.to_string();
        let subject = subject.to_string();
        self.with_conn(move |conn| {
            conn.query_row(
                "SELECT client_id
                 FROM upstream_oauth_dynamic_clients
                 WHERE upstream_name = ?1 AND subject = ?2",
                params![upstream_name, subject],
                |row| row.get(0),
            )
            .optional()
            .map_err(sqlite_error)
        })
        .await
    }

    pub async fn delete_dynamic_client_registration(
        &self,
        upstream_name: &str,
        subject: &str,
    ) -> Result<(), AuthError> {
        let upstream_name = upstream_name.to_string();
        let subject = subject.to_string();
        self.with_conn(move |conn| {
            conn.execute(
                "DELETE FROM upstream_oauth_dynamic_clients
                 WHERE upstream_name = ?1 AND subject = ?2",
                params![upstream_name, subject],
            )
            .map_err(sqlite_error)?;
            Ok(())
        })
        .await
    }

    /// Add an email address to the allowlist.
    ///
    /// `email` is normalised to lowercase before storage. Returns
    /// `AuthError::Validation` if the email is already present.
    pub async fn add_allowed_user(
        &self,
        email: &str,
        added_by: &str,
        created_at: i64,
    ) -> Result<(), AuthError> {
        let email = email.to_lowercase();
        let fp = fingerprint(&email);
        let added_by = added_by.to_string();
        self.with_conn(move |conn| {
            let changed = conn
                .execute(
                    "INSERT INTO allowed_users (email, added_by, created_at)
                     VALUES (?1, ?2, ?3)",
                    params![email, added_by, created_at],
                )
                .map_err(|error| match error {
                    rusqlite::Error::SqliteFailure(ref e, _)
                        if e.code == rusqlite::ErrorCode::ConstraintViolation =>
                    {
                        AuthError::Validation(format!(
                            "email fingerprint {fp} is already in the allowlist"
                        ))
                    }
                    other => sqlite_error(other),
                })?;
            debug_assert_eq!(changed, 1);
            Ok(())
        })
        .await
    }

    /// Remove an email address from the allowlist.
    ///
    /// Idempotent: returns `Ok(())` even if the email was not present.
    pub async fn remove_allowed_user(&self, email: &str) -> Result<(), AuthError> {
        let email = email.to_lowercase();
        self.with_conn(move |conn| {
            conn.execute("DELETE FROM allowed_users WHERE email = ?1", params![email])
                .map_err(sqlite_error)?;
            Ok(())
        })
        .await
    }

    /// Return all allowlist rows ordered by `created_at ASC`.
    pub async fn list_allowed_users(&self) -> Result<Vec<AllowedUserRow>, AuthError> {
        self.with_conn(move |conn| {
            let mut stmt = conn
                .prepare(
                    "SELECT email, added_by, created_at
                     FROM allowed_users
                     ORDER BY created_at ASC",
                )
                .map_err(sqlite_error)?;
            let rows = stmt
                .query_map([], row_to_allowed_user)
                .map_err(sqlite_error)?
                .collect::<rusqlite::Result<Vec<_>>>()
                .map_err(sqlite_error)?;
            Ok(rows)
        })
        .await
    }

    async fn with_conn<T, F>(&self, op: F) -> Result<T, AuthError>
    where
        T: Send + 'static,
        F: FnOnce(&Connection) -> Result<T, AuthError> + Send + 'static,
    {
        let conns = Arc::clone(&self.conns);
        let path = Arc::clone(&self.path);
        let len = conns.len();
        let idx = self.next_conn.fetch_add(1, Ordering::Relaxed) % len;
        tokio::task::spawn_blocking(move || {
            let mut guard = conns[idx]
                .lock()
                .map_err(|_| AuthError::Storage("sqlite mutex poisoned".to_string()))?;
            validate_or_reopen_connection(&mut guard, path.as_ref())?;
            op(&guard)
        })
        .await
        .map_err(|error| AuthError::Storage(format!("sqlite task failed: {error}")))?
    }

    #[cfg(test)]
    fn connection_count(&self) -> usize {
        self.conns.len()
    }
}

fn open_connections(path: &Path, count: usize) -> Result<Vec<Connection>, AuthError> {
    (0..count).map(|_| open_connection(path)).collect()
}

#[allow(clippy::too_many_lines)]
fn open_connection(path: &Path) -> Result<Connection, AuthError> {
    if let Some(parent) = path.parent() {
        std::fs::create_dir_all(parent).map_err(|error| {
            AuthError::Storage(format!(
                "create auth database directory `{}`: {error}",
                parent.display()
            ))
        })?;
    }

    let existed = path.exists();
    if existed {
        ensure_restrictive_permissions(path)?;
    }

    let conn = Connection::open(path).map_err(sqlite_error)?;
    conn.busy_timeout(std::time::Duration::from_millis(SQLITE_BUSY_TIMEOUT_MS))
        .map_err(sqlite_error)?;
    conn.pragma_update(None, "journal_mode", "WAL")
        .map_err(sqlite_error)?;
    conn.pragma_update(None, "foreign_keys", "ON")
        .map_err(sqlite_error)?;
    conn.execute_batch(
        "CREATE TABLE IF NOT EXISTS registered_clients (
            client_id TEXT PRIMARY KEY,
            redirect_uris TEXT NOT NULL,
            created_at INTEGER NOT NULL
        );
        CREATE TABLE IF NOT EXISTS authorization_requests (
            state TEXT PRIMARY KEY,
            client_id TEXT NOT NULL,
            redirect_uri TEXT NOT NULL,
            client_state TEXT NOT NULL,
            resource TEXT NOT NULL DEFAULT '',
            scope TEXT NOT NULL,
            provider TEXT NOT NULL DEFAULT 'google',
            provider_code_verifier TEXT NOT NULL,
            code_challenge TEXT NOT NULL,
            code_challenge_method TEXT NOT NULL,
            created_at INTEGER NOT NULL,
            expires_at INTEGER NOT NULL
        );
        CREATE TABLE IF NOT EXISTS authorization_codes (
            code TEXT PRIMARY KEY,
            client_id TEXT NOT NULL,
            subject TEXT NOT NULL,
            redirect_uri TEXT NOT NULL,
            resource TEXT NOT NULL DEFAULT '',
            scope TEXT NOT NULL,
            provider TEXT NOT NULL DEFAULT 'google',
            code_challenge TEXT NOT NULL,
            code_challenge_method TEXT NOT NULL,
            provider_refresh_token TEXT,
            created_at INTEGER NOT NULL,
            expires_at INTEGER NOT NULL
        );
        CREATE TABLE IF NOT EXISTS refresh_tokens (
            refresh_token_hash TEXT PRIMARY KEY,
            client_id TEXT NOT NULL,
            subject TEXT NOT NULL,
            resource TEXT NOT NULL DEFAULT '',
            scope TEXT NOT NULL,
            provider TEXT NOT NULL DEFAULT 'google',
            provider_refresh_token TEXT,
            created_at INTEGER NOT NULL,
            expires_at INTEGER NOT NULL
        );
        CREATE TABLE IF NOT EXISTS browser_sessions (
            session_id TEXT PRIMARY KEY,
            subject TEXT NOT NULL,
            email TEXT,
            csrf_token TEXT NOT NULL,
            created_at INTEGER NOT NULL,
            expires_at INTEGER NOT NULL
        );
        CREATE TABLE IF NOT EXISTS browser_login_states (
            state TEXT PRIMARY KEY,
            return_to TEXT NOT NULL,
            provider TEXT NOT NULL DEFAULT 'google',
            provider_code_verifier TEXT NOT NULL,
            created_at INTEGER NOT NULL,
            expires_at INTEGER NOT NULL
        );
        CREATE TABLE IF NOT EXISTS native_authorization_results (
            state TEXT PRIMARY KEY,
            code TEXT NOT NULL,
            created_at INTEGER NOT NULL,
            expires_at INTEGER NOT NULL
        );
        CREATE TABLE IF NOT EXISTS upstream_oauth_credentials (
            upstream_name             TEXT NOT NULL,
            subject                   TEXT NOT NULL,
            client_id                 TEXT NOT NULL,
            granted_scopes_json       TEXT NOT NULL,
            token_blob                BLOB NOT NULL,
            token_blob_nonce          BLOB NOT NULL,
            token_received_at         INTEGER NOT NULL,
            access_token_expires_at   INTEGER NOT NULL,
            refresh_token_present     INTEGER NOT NULL,
            PRIMARY KEY (upstream_name, subject)
        ) WITHOUT ROWID;
        CREATE TABLE IF NOT EXISTS upstream_oauth_state (
            upstream_name   TEXT NOT NULL,
            subject         TEXT NOT NULL,
            csrf_token      TEXT NOT NULL,
            pkce_verifier   TEXT NOT NULL,
            created_at      INTEGER NOT NULL,
            expires_at      INTEGER NOT NULL,
            PRIMARY KEY (upstream_name, subject, csrf_token)
        ) WITHOUT ROWID;
        CREATE TABLE IF NOT EXISTS upstream_oauth_dynamic_clients (
            upstream_name   TEXT NOT NULL,
            subject         TEXT NOT NULL,
            client_id       TEXT NOT NULL,
            created_at      INTEGER NOT NULL,
            PRIMARY KEY (upstream_name, subject)
        ) WITHOUT ROWID;
        CREATE TABLE IF NOT EXISTS allowed_users (
            email       TEXT PRIMARY KEY NOT NULL,
            added_by    TEXT NOT NULL,
            created_at  INTEGER NOT NULL
        );",
    )
    .map_err(sqlite_error)?;
    add_column_if_missing(
        &conn,
        "authorization_requests",
        "resource",
        "TEXT NOT NULL DEFAULT ''",
    )?;
    add_column_if_missing(
        &conn,
        "authorization_codes",
        "resource",
        "TEXT NOT NULL DEFAULT ''",
    )?;
    add_column_if_missing(
        &conn,
        "refresh_tokens",
        "resource",
        "TEXT NOT NULL DEFAULT ''",
    )?;
    add_column_if_missing(
        &conn,
        "authorization_requests",
        "provider",
        "TEXT NOT NULL DEFAULT 'google'",
    )?;
    add_column_if_missing(
        &conn,
        "authorization_codes",
        "provider",
        "TEXT NOT NULL DEFAULT 'google'",
    )?;
    add_column_if_missing(
        &conn,
        "refresh_tokens",
        "provider",
        "TEXT NOT NULL DEFAULT 'google'",
    )?;
    add_column_if_missing(
        &conn,
        "browser_login_states",
        "provider",
        "TEXT NOT NULL DEFAULT 'google'",
    )?;

    if !existed {
        set_restrictive_permissions(path)?;
    }
    ensure_restrictive_permissions(path)?;

    run_migrations(&conn)?;

    Ok(conn)
}

/// One-time migrations keyed by `PRAGMA user_version`.
///
/// Migration 0 → 1: add `refresh_token_hash` to the `refresh_tokens` table
/// (if the table was created with the old `refresh_token TEXT PRIMARY KEY`
/// schema) and backfill SHA-256 hashes for any plaintext rows that pre-date
/// this change.  New databases created with the v1 schema already have
/// `refresh_token_hash` as the PK, so the `ALTER TABLE` step is a no-op in
/// that case.
fn run_migrations(conn: &Connection) -> Result<(), AuthError> {
    let current_version: i64 = conn
        .query_row("PRAGMA user_version;", [], |row| row.get(0))
        .map_err(sqlite_error)?;

    if current_version < 1 {
        // Step 1: add `refresh_token_hash` column if missing (pre-v1 DBs have
        // `refresh_token TEXT PRIMARY KEY` and no hash column).
        let cols: Vec<String> = {
            let mut stmt = conn
                .prepare("PRAGMA table_info(refresh_tokens);")
                .map_err(sqlite_error)?;
            stmt.query_map([], |row| row.get::<_, String>(1))
                .map_err(sqlite_error)?
                .collect::<rusqlite::Result<Vec<_>>>()
                .map_err(sqlite_error)?
        };

        if !cols.iter().any(|c| c == "refresh_token_hash") {
            // Old schema: add the column and back-fill SHA-256 hashes.
            conn.execute_batch("ALTER TABLE refresh_tokens ADD COLUMN refresh_token_hash TEXT;")
                .map_err(sqlite_error)?;

            // Back-fill: hash existing plaintext `refresh_token` values.  We
            // can only do this in a SQL-only migration when the hash is
            // computed outside SQLite; instead load all rows, compute hashes
            // in Rust, and update.
            let rows: Vec<(String,)> = {
                let mut stmt = conn
                    .prepare("SELECT refresh_token FROM refresh_tokens WHERE refresh_token_hash IS NULL;")
                    .map_err(sqlite_error)?;
                stmt.query_map([], |row| Ok((row.get::<_, String>(0)?,)))
                    .map_err(sqlite_error)?
                    .collect::<rusqlite::Result<Vec<_>>>()
                    .map_err(sqlite_error)?
            };
            for (plaintext,) in rows {
                let hash = hash_token(&plaintext);
                conn.execute(
                    "UPDATE refresh_tokens SET refresh_token_hash = ?1 WHERE refresh_token = ?2 AND refresh_token_hash IS NULL;",
                    params![hash, plaintext],
                )
                .map_err(sqlite_error)?;
            }

            warn!(
                "migration v1: added refresh_token_hash column and backfilled existing rows — old plaintext tokens invalidated on next rotation"
            );
        }

        // Ensure a UNIQUE index exists on refresh_token_hash so that
        // ON CONFLICT(refresh_token_hash) works correctly on pre-existing
        // databases where the column was added by ALTER TABLE (not declared as
        // PRIMARY KEY).  On new databases the column is already PRIMARY KEY so
        // this index is redundant but harmless.
        conn.execute_batch(
            "CREATE UNIQUE INDEX IF NOT EXISTS idx_refresh_tokens_hash \
             ON refresh_tokens(refresh_token_hash);",
        )
        .map_err(sqlite_error)?;

        conn.execute_batch("PRAGMA user_version = 1;")
            .map_err(sqlite_error)?;
    }

    if current_version < 2 {
        // Step 2: add `dynamic_client_id` column to `upstream_oauth_state`.
        // This column binds the OAuth client_id used to begin a specific
        // authorization flow to the CSRF state row so that concurrent
        // `begin_authorization` calls for the same upstream+subject can each
        // complete their own callback with the correct client_id (lab-77y5.15).
        add_column_if_missing(conn, "upstream_oauth_state", "dynamic_client_id", "TEXT")?;

        conn.execute_batch(&format!("PRAGMA user_version = {SCHEMA_VERSION};"))
            .map_err(sqlite_error)?;
    }

    Ok(())
}

/// Compute a hex-encoded SHA-256 digest of a token for safe storage.
///
/// The raw token (24+ bytes of random entropy) has sufficient pre-image
/// resistance for SHA-256 to be appropriate here — Argon2 would add
/// per-request latency without a meaningful security benefit.
/// AAD binding an encrypted `provider_refresh_token` to its row identity.
///
/// The refresh-token hash is the table's primary key and is derivable at
/// both encrypt time (upsert/rotate hash the plaintext token before insert)
/// and decrypt time (lookup is keyed by the same hash), so a ciphertext
/// copied onto a row with a different hash fails authentication.  Mirrors
/// the `key=value` AAD shape used by `upstream::store::credential_aad`.
fn refresh_token_aad(refresh_token_hash: &str) -> Vec<u8> {
    format!("refresh_token_hash={refresh_token_hash}").into_bytes()
}

fn hash_token(token: &str) -> String {
    let digest = Sha256::digest(token.as_bytes());
    let mut hex = String::with_capacity(64);
    for byte in &digest {
        let _ = write!(&mut hex, "{byte:02x}");
    }
    hex
}

fn validate_or_reopen_connection(conn: &mut Connection, path: &Path) -> Result<(), AuthError> {
    let Err(error) = conn.query_row("SELECT 1", [], |row| row.get::<_, i64>(0)) else {
        return Ok(());
    };
    warn!(
        path = %path.display(),
        error = %error,
        "stale sqlite connection detected, reopening"
    );

    *conn = open_connection(path)?;
    conn.query_row("SELECT 1", [], |row| row.get::<_, i64>(0))
        .map(|_| ())
        .map_err(sqlite_error)
}

#[allow(clippy::needless_pass_by_value)]
fn sqlite_error(error: rusqlite::Error) -> AuthError {
    AuthError::Storage(format!("sqlite error: {error}"))
}

fn add_column_if_missing(
    conn: &Connection,
    table: &str,
    column: &str,
    definition: &str,
) -> Result<(), AuthError> {
    let mut stmt = conn
        .prepare(&format!("PRAGMA table_info({table})"))
        .map_err(sqlite_error)?;
    let exists = stmt
        .query_map([], |row| row.get::<_, String>(1))
        .map_err(sqlite_error)?
        .collect::<rusqlite::Result<Vec<_>>>()
        .map_err(sqlite_error)?
        .iter()
        .any(|name| name == column);
    if !exists {
        conn.execute(
            &format!("ALTER TABLE {table} ADD COLUMN {column} {definition}"),
            [],
        )
        .map_err(sqlite_error)?;
    }
    Ok(())
}

#[cfg(test)]
#[path = "sqlite_tests.rs"]
mod tests;

#[cfg(test)]
#[path = "sqlite_migration_tests.rs"]
mod migration_tests;