db_keystore/

lib.rs

//! File-backed credential store using Turso (sqlite) and optional encryption.
//!
//! This module implements the `keyring_core::api::CredentialStoreApi` and
//! `keyring_core::api::CredentialApi` traits, so it can be used wherever a
//! `keyring_core::api::CredentialStore` is expected (for example via
//! `use_named_store_with_modifiers`).
//!
//! Features:
//! - Local sqlite storage with optional encryption options.
//! - Verified out-of-place rekey (DEK rotation) with safe destination
//!   creation and typed errors, plus standalone verification of two existing
//!   keystores ([`DbKeyStore::verify`]): see the [`rekey`] module.
//! - WAL + busy timeout for better multi-process behavior.
//! - Multi-process file sharing: the file-backed store opens the database
//!   per operation and closes it immediately afterward, so turso's
//!   exclusive file lock is held only for the duration of each operation.
//!   Long-lived processes (daemons) and short-lived CLI invocations can
//!   share one keystore file; concurrent opens are handled by retry with
//!   backoff.
//! - Optional uniqueness enforcement on (service, user) via `allow_ambiguity=false`.
//! - UUID and optional comment attributes exposed via the credential API.
//! - Search supports `service`, `user`, `uuid`, and `comment` regex filters.
//!
//! Modifiers supported by `new_with_modifiers`:
//! - `path` : path to the sqlite database file. Defaults to `$XDG_STATE_HOME/keystore.db` or `$HOME/.local/state/keystore.db`
//! - `encryption-cipher` / `cipher`: encryption cipher name (optional, requires hexkey).
//! - `encryption-hexkey` / `hexkey`: encryption key as hex (optional, requires cipher).
//! - `allow-ambiguity` / `allow_ambiguity`: `"true"` or `"false"` (default `"false"`).
//! - `vfs`: optional VFS backing selection (`"memory"`, `"io_uring"`, or `"syscall"`).
//! - `index-always` / `index_always`: `"true"` or `"false"` (default `"false"`).
//!
//! Modifiers supported by `build`:
//! - `uuid`: explicit credential UUID (allows creating ambiguous entries when allowed).
//! - `comment`: initial comment value stored with the credential.
//!
//! Uuid are generated in v7 format <https://www.ietf.org/rfc/rfc9562.html#section-5.7>.
//! Uuids generated by this crate will be unique (on a per-process basis), and sortable by time,
//! so ambiguous entries can be sorted by date created, if desired. Uuids generated externally,
//! and passed to `build()` are validated against the string syntax
//! (e.g., `f81d4fae-7dec-11d0-a765-00a0c91e6bf6`), but are not checked for uniqueness or order.
//!
//!
//! Example:
//! ```rust
//! use std::collections::HashMap;
//! use db_keystore::{DbKeyStore, DbKeyStoreConfig};
//!
//! // create from config
//! let config = DbKeyStoreConfig {
//!     path: "keystore.db".into(),
//!     ..Default::default()
//! };
//! let store = DbKeyStore::new(config).expect("store");
//!
//! // or, create with modifiers
//! let modifiers = HashMap::from([
//!     ("path", "keystore.db"),
//!     ("allow-ambiguity", "true"),
//! ]);
//! let store = DbKeyStore::new_with_modifiers(&modifiers).expect("store");
//! ```
#![warn(clippy::pedantic)]
#![allow(clippy::missing_errors_doc)]
#![allow(clippy::must_use_candidate)]

// SAFETY - Security and safety notes:
//  - SQL injection: all user data is bound as parameters; SQL is static.
//  - Secret handling: secrets are validated with length checks.
//    Optional on-disk encryption implemented in database.
//  - Concurrency: set_secret uses a transaction for read/modify/write; single statements
//    are atomic in sqlite.
//  - Contention: connections enable WAL and busy_timeout to reduce sqlite_BUSY in
//    multi-process usage. File-backed databases are opened per operation and
//    closed immediately after (see `Backend`), so turso's exclusive file lock
//    is never held between operations; concurrent opens retry with backoff.
//  - Uniqueness: allow_ambiguity=false enforces a unique (service,user) index and
//    uses UPSERT; allow_ambiguity=true permits multiple credentials per pair.
//  - Zeroize used to prevent secrets (db encryption keys and keyring secrets)
//    leaking into heap from this crate.
use std::{
    collections::HashMap,
    fmt,
    path::{Path, PathBuf},
    sync::Arc,
    time::{SystemTime, UNIX_EPOCH},
};

use futures::executor::block_on;
use keyring_core::{
    api::{CredentialApi, CredentialPersistence, CredentialStoreApi},
    attributes::parse_attributes,
    {Credential, Entry, Error, Result},
};
use regex::Regex;
use turso::{Builder, Connection, Database, Value};
use zeroize::Zeroizing;

pub mod rekey;
pub use rekey::{RekeyError, RekeyOutcome, SensitiveKey};
#[cfg(target_os = "linux")]
pub use rekey::{rekey_at, verify_at};

const CRATE_VERSION: &str = env!("CARGO_PKG_VERSION");

// length limits to prevent accidental blow up of db:
//  - service and name: 1024 bytes
//  - secret: 65536 bytes
const MAX_NAME_LEN: u32 = 1024;
const MAX_SECRET_LEN: u32 = 65536;
const SCHEMA_VERSION: u32 = 1;
// sqlite timeout for connection busy
const BUSY_TIMEOUT_MS: u32 = 5000;
/// retry logic for open and connect, in case there's a temporary file lock
const OPEN_LOCK_RETRIES: u32 = 60;
const OPEN_LOCK_BACKOFF_MS: u64 = 20;
const OPEN_LOCK_BACKOFF_MAX_MS: u64 = 250;

/// `EncryptionOpts` selects the cipher and key for on-disk encryption.
/// See <https://docs.turso.tech/tursodb/encryption>
/// Example ciphers: "aegis256", "aes256gcm". For 256-bit keys, hexkey is 64 chars.
///
/// The key is decoded from hex at construction and held in a fixed-size
/// zeroizing buffer ([`SensitiveKey`]) that is wiped when dropped. The
/// constructor borrows the caller's hex string; the caller stays responsible
/// for wiping its own copy.
#[derive(Clone)]
pub struct EncryptionOpts {
    cipher: String,
    key: SensitiveKey,
}

impl EncryptionOpts {
    /// Create encryption options from a cipher name and a hex-encoded key.
    ///
    /// The key is copied into zeroizing storage; the borrowed `hexkey` is left
    /// untouched (callers should wipe their own copy, e.g. with `zeroize`).
    /// Returns an error if the hex is malformed or its length does not match
    /// the cipher's key size.
    pub fn new(cipher: &str, hexkey: &str) -> Result<Self> {
        let key = SensitiveKey::from_hex(hexkey)
            .map_err(|e| Error::Invalid("hexkey".to_string(), e.to_string()))?;
        Self::with_key(cipher, key)
    }

    /// Create encryption options from a cipher name and an already-decoded key.
    pub fn with_key(cipher: &str, key: SensitiveKey) -> Result<Self> {
        if cipher.is_empty() {
            return Err(Error::Invalid(
                "cipher".to_string(),
                "cipher must not be empty".to_string(),
            ));
        }
        if let Some(expected) = cipher_key_len(cipher)
            && expected != key.len()
        {
            return Err(Error::Invalid(
                "hexkey".to_string(),
                format!(
                    "cipher '{cipher}' requires a {expected}-byte key ({} hex chars)",
                    expected * 2
                ),
            ));
        }
        Ok(Self {
            cipher: cipher.to_string(),
            key,
        })
    }

    /// Cipher name.
    pub fn cipher(&self) -> &str {
        &self.cipher
    }

    /// Borrow the raw key bytes. The bytes remain owned by zeroizing storage.
    pub fn key_bytes(&self) -> &[u8] {
        self.key.as_bytes()
    }

    /// Hex-encode the key into a zeroizing buffer for the turso boundary.
    pub(crate) fn key_hex(&self) -> Zeroizing<String> {
        self.key.to_hex()
    }
}

impl fmt::Debug for EncryptionOpts {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("EncryptionOpts")
            .field("cipher", &self.cipher)
            .field("key", &"<redacted>")
            .finish()
    }
}

/// Key size in bytes for known turso cipher names (dash/underscore aliases
/// accepted). Unknown ciphers return `None` and are passed through to turso.
pub(crate) fn cipher_key_len(cipher: &str) -> Option<usize> {
    let normalized: String = cipher
        .to_ascii_lowercase()
        .chars()
        .filter(|c| *c != '-' && *c != '_')
        .collect();
    match normalized.as_str() {
        "aegis128l" | "aegis128x2" | "aegis128x4" | "aes128gcm" => Some(16),
        "aegis256" | "aegis256x2" | "aegis256x4" | "aes256gcm" => Some(32),
        _ => None,
    }
}

/// Build the `turso::EncryptionOpts` handed to the database builder.
///
/// NOTE (turso boundary): `turso::EncryptionOpts` stores the key as an
/// ordinary `String`, which turso frees without wiping. Everything on the
/// db-keystore side of this call is zeroizing; extending the guarantee into
/// turso requires a turso API change (out of scope here).
pub(crate) fn turso_encryption_opts(opts: &EncryptionOpts) -> turso::EncryptionOpts {
    let hexkey = opts.key_hex();
    turso::EncryptionOpts {
        cipher: opts.cipher().to_string(),
        hexkey: hexkey.as_str().to_string(),
    }
}

/// Generates a new unique uuid as a string.
/// v7 format: 48 bit timestamp in milliseconds and 78 bits of randomness
/// `https://www.ietf.org/rfc/rfc9562.html#section-5.7`
fn new_uuid() -> String {
    uuid::Uuid::now_v7().to_string()
}

/// Configure turso database
#[derive(Debug, Default, Clone)]
pub struct DbKeyStoreConfig {
    /// Path to database. Defaults to `$XDG_STATE_HOME/keystore.db` or `$HOME/.local/state/keystore.db`
    pub path: PathBuf,

    /// Set cipher and encryption key to enable encryption
    pub encryption_opts: Option<EncryptionOpts>,

    /// Allow non-unique values for (service,user) (see keyring-core documentation)
    pub allow_ambiguity: bool,

    /// Database I/O strategy: "`memory`", "`syscall`", or "`io_uring`"
    ///  - "`memory`": In-memory database. Data is entirely in RAM, and data is lost when process exits. When vfs=memory, `path` and `encryption_opts` are ignored.
    ///  - "`syscall`": Generic syscall backend. Uses standard POSIX system calls for file I/O. This is the most portable mode.
    ///  - "`io_uring"`: Linux `io_uring` backend. Uses Linux's modern async I/O interface for better performance. Only available on Linux.
    pub vfs: Option<String>,

    /// Add index on (service,user) even when `allow_ambiguity` is true.
    /// Increases file size about 2x, improves performance for large keystores (>~500 entries)
    pub index_always: bool,
}

/// Default path for keystore: `$XDG_STATE_HOME/keystore.db` or `$HOME/.local/state/keystore.db`
pub fn default_path() -> Result<PathBuf> {
    Ok(match std::env::var("XDG_STATE_HOME") {
        Ok(dir) => PathBuf::from(dir),
        _ => match std::env::var("HOME") {
            Ok(home) => PathBuf::from(home).join(".local").join("state"),
            _ => {
                return Err(Error::Invalid(
                    "path".to_owned(),
                    "No default path: set 'path' in Config (or modifiers), or define XDG_STATE_HOME or HOME"
                        .to_owned(),
                ));
            }
        },
    }
    .join("keystore.db"))
}

#[derive(Clone)]
pub struct DbKeyStore {
    inner: Arc<DbKeyStoreInner>,
}

#[derive(Debug)]
struct DbKeyStoreInner {
    backend: Backend,
    id: String,
    allow_ambiguity: bool,
    encrypted: bool,
    path: String,
}

/// How the store reaches its database.
///
/// turso holds an exclusive `fcntl` lock on a file-backed database for as
/// long as the `Database` handle is open — even for reads — so only one
/// process system-wide can have the file open at a time. To let multiple
/// processes share one keystore file, the file backend stores the open
/// parameters and reopens the database per operation, dropping the handle
/// (and with it the lock) as soon as the operation completes. Overlapping
/// opens from other processes are absorbed by the retry/backoff in
/// [`open_db_with_retry`]. The in-memory backend keeps its database open
/// for the store's lifetime, since its contents live only in the handle.
#[derive(Debug)]
enum Backend {
    /// Persistent in-memory database, held open for the store's lifetime.
    Memory(Database),
    /// File-backed database, reopened per operation from these parameters
    /// (the path lives in [`DbKeyStoreInner::path`]).
    File {
        encryption_opts: Option<EncryptionOpts>,
        vfs: Option<String>,
    },
}

/// One open database session, yielded by [`DbKeyStoreInner::connect`].
///
/// For the file backend this owns the reopened [`Database`]; dropping the
/// session closes the file and releases turso's exclusive lock. Derefs to
/// [`Connection`] so call sites use it wherever a connection is expected.
struct DbSession {
    /// Keeps a file-backed database open for the life of the session;
    /// `None` for the persistent in-memory backend.
    db: Option<Database>,
    conn: Connection,
}

impl std::ops::Deref for DbSession {
    type Target = Connection;
    fn deref(&self) -> &Connection {
        &self.conn
    }
}

impl Drop for DbSession {
    fn drop(&mut self) {
        if self.db.is_some() {
            checkpoint_before_close(&self.conn);
        }
    }
}

/// Folds the WAL into the main database file before a file-backed handle
/// closes.
///
/// turso does not coordinate WAL state across processes (its multi-process
/// WAL support is experimental and off here), so a WAL left behind at close
/// is not reliably replayed when the next opener is a different process —
/// writes can appear to vanish ("no such table: credentials"). Truncating
/// the WAL while we still hold the exclusive file lock guarantees the next
/// opener starts from a complete main file and an empty WAL. Best-effort:
/// an error leaves a non-empty WAL for same-process reopen recovery, which
/// is the pre-existing behavior.
fn checkpoint_before_close(conn: &Connection) {
    let result = block_on(async {
        let mut rows = conn.query("PRAGMA wal_checkpoint(TRUNCATE)", ()).await?;
        while (rows.next().await?).is_some() {}
        Ok::<(), turso::Error>(())
    });
    if let Err(err) = result {
        log::debug!("wal_checkpoint(TRUNCATE) on close failed: {err}");
    }
}

#[derive(Debug, Clone, Eq, PartialEq, Hash)]
struct CredId {
    service: String,
    user: String,
}

#[derive(Debug, Clone)]
struct DbKeyCredential {
    inner: Arc<DbKeyStoreInner>,
    id: CredId,
    uuid: Option<String>,
    comment: Option<String>,
}

#[derive(Debug)]
enum LookupResult<T> {
    None,
    One(T),
    Ambiguous(Vec<String>),
}

#[derive(Debug)]
struct CommentRow {
    uuid: String,
    comment: Option<String>,
}

impl DbKeyStore {
    pub fn new(config: DbKeyStoreConfig) -> Result<Arc<DbKeyStore>> {
        let start_time = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap_or_default()
            .as_secs_f64();
        // move the (zeroizing) key out of the config so it has a single owner
        let encryption_opts = config.encryption_opts;
        if let Some(vfs) = &config.vfs
            && vfs == "memory"
        {
            // in-memory database. ignore path and encryption options
            let db = map_turso(block_on(async {
                Builder::new_local(":memory:")
                    .with_io("memory".into())
                    .build()
                    .await
            }))?;
            let id = format!("DbKeyStore v{CRATE_VERSION} in-memory @ {start_time}");
            let conn = map_turso(db.connect())?;
            init_schema(&conn, config.allow_ambiguity, config.index_always)?;
            return Ok(Arc::new(DbKeyStore {
                inner: Arc::new(DbKeyStoreInner {
                    backend: Backend::Memory(db),
                    id,
                    allow_ambiguity: config.allow_ambiguity,
                    encrypted: false,
                    path: ":memory:".to_string(),
                }),
            }));
        }
        let path = if config.path.as_os_str().is_empty() {
            default_path()?
        } else {
            config.path.clone()
        };
        // turso requires paths to be valid utf8
        let path_str = path
            .to_str()
            .ok_or_else(|| Error::Invalid("path".into(), "path must be valid UTF-8".to_string()))?;
        ensure_parent_dir(&path)?;
        let encrypted = encryption_opts.is_some();
        // Open once to validate the configuration (a wrong encryption key
        // fails here) and initialize the schema. The handle is dropped at the
        // end of this block so the exclusive file lock is released; each
        // subsequent operation reopens the database via `connect`.
        {
            let db = open_db_with_retry(path_str, encryption_opts.as_ref(), config.vfs.as_deref())?;
            let conn = retry_turso_locking(|| db.connect())?;
            configure_connection(&conn)?;
            init_schema(&conn, config.allow_ambiguity, config.index_always)?;
            checkpoint_before_close(&conn);
        }
        let id =
            format!("DbKeyStore v{CRATE_VERSION} path:{path_str} enc:{encrypted} @ {start_time}");
        Ok(Arc::new(DbKeyStore {
            inner: Arc::new(DbKeyStoreInner {
                backend: Backend::File {
                    encryption_opts,
                    vfs: config.vfs.clone(),
                },
                id,
                allow_ambiguity: config.allow_ambiguity,
                encrypted,
                path: path_str.to_string(),
            }),
        }))
    }

    pub fn new_with_modifiers(modifiers: &HashMap<&str, &str>) -> Result<Arc<DbKeyStore>> {
        // map is mutable so we can move hexkey into Zeroize and avoid dropping the String
        let mut mods = parse_attributes(
            &[
                "path",
                "encryption-cipher",
                "cipher",
                "encryption-hexkey",
                "hexkey",
                "*allow-ambiguity",
                "*allow_ambiguity",
                "vfs",
                "*index-always",
                "*index_always",
            ],
            Some(modifiers),
        )?;
        let path = mods.remove("path").map(PathBuf::from).unwrap_or_default();
        let cipher = mods
            .remove("encryption-cipher")
            .or_else(|| mods.remove("cipher"));
        // keep our copy of the hex key in a zeroizing owner
        let hexkey = mods
            .remove("encryption-hexkey")
            .or_else(|| mods.remove("hexkey"))
            .map(Zeroizing::new);
        let allow_ambiguity = mods
            .remove("allow-ambiguity")
            .or_else(|| mods.remove("allow_ambiguity"))
            .is_some_and(|value| value == "true");
        let index_always = mods
            .remove("index-always")
            .or_else(|| mods.remove("index_always"))
            .is_some_and(|value| value == "true");
        let vfs = mods.remove("vfs");
        let encryption_opts = match (cipher, hexkey) {
            (None, None) => None,
            (Some(cipher), Some(hexkey)) => Some(EncryptionOpts::new(&cipher, hexkey.as_str())?),
            _ => {
                return Err(Error::Invalid(
                    "encryption".to_string(),
                    "encryption-cipher and encryption-hexkey must both be set".to_string(),
                ));
            }
        };
        let config = DbKeyStoreConfig {
            path,
            encryption_opts,
            allow_ambiguity,
            vfs,
            index_always,
        };
        DbKeyStore::new(config)
    }

    /// Returns true if the db file is encrypted
    pub fn is_encrypted(&self) -> bool {
        self.inner.encrypted
    }

    /// Returns path to database file
    pub fn path(&self) -> String {
        self.inner.path.clone()
    }
}

impl std::fmt::Debug for DbKeyStore {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("DbKeyStore")
            .field("vendor", &self.vendor())
            .field("id", &self.id())
            .field("allow_ambiguity", &self.inner.allow_ambiguity)
            .finish()
    }
}

impl DbKeyStoreInner {
    /// Opens a database session for one operation.
    ///
    /// The file backend reopens the database here and the returned
    /// [`DbSession`] owns the handle, so turso's exclusive file lock is
    /// held only until the session is dropped.
    fn connect(&self) -> Result<DbSession> {
        match &self.backend {
            Backend::Memory(db) => {
                let conn = map_turso(db.connect())?;
                configure_connection(&conn)?;
                Ok(DbSession { db: None, conn })
            }
            Backend::File {
                encryption_opts,
                vfs,
            } => {
                let db = open_db_with_retry(&self.path, encryption_opts.as_ref(), vfs.as_deref())?;
                let conn = retry_turso_locking(|| db.connect())?;
                configure_connection(&conn)?;
                Ok(DbSession { db: Some(db), conn })
            }
        }
    }
}

impl DbKeyCredential {
    async fn insert_credential(
        &self,
        conn: &Connection,
        uuid: &str,
        secret: Value,
        comment: Value,
    ) -> Result<()> {
        conn.execute(
            "INSERT INTO credentials (service, user, uuid, secret, comment) VALUES (?1, ?2, ?3, ?4, ?5)",
            (
                self.id.service.as_str(),
                self.id.user.as_str(),
                uuid,
                secret,
                comment,
            ),
        )
        .await
        .map_err(map_turso_err)?;
        Ok(())
    }
}

impl CredentialStoreApi for DbKeyStore {
    fn vendor(&self) -> String {
        String::from("DbKeyStore, https://crates.io/crates/db-keystore")
    }

    fn id(&self) -> String {
        self.inner.id.clone()
    }

    /// Create a credential entry for service and user.
    /// Service and user must be non-empty, and within the length limits. (<=1024 chars)
    /// Supported modifiers: `uuid`, `comment`.
    fn build(
        &self,
        service: &str,
        user: &str,
        modifiers: Option<&HashMap<&str, &str>>,
    ) -> Result<Entry> {
        validate_service_user(service, user)?;
        let mods = parse_attributes(&["uuid", "comment"], modifiers)?;
        let credential = DbKeyCredential {
            inner: Arc::clone(&self.inner),
            id: CredId {
                service: service.to_string(),
                user: user.to_string(),
            },
            uuid: mods
                .get("uuid")
                .map(|value| normalize_uuid_input(value))
                .transpose()?,
            comment: mods.get("comment").cloned(),
        };
        Ok(Entry::new_with_credential(Arc::new(credential)))
    }

    // Search based on regex criteria, returning a list of matching entries.
    // include any of "service", "user", "uuid", or "comment" as (regex) search terms
    // Notes:
    // - uuids in the database are lowercase
    // - If "comment" is an empty string, it matches entries with no comment.
    //   To match on "any" comment, omit comment from the search spec.
    fn search(&self, spec: &HashMap<&str, &str>) -> Result<Vec<Entry>> {
        let spec = parse_attributes(&["service", "user", "uuid", "comment"], Some(spec))?;
        let service_re = Regex::new(spec.get("service").map_or("", String::as_str))
            .map_err(|e| Error::Invalid("service regex".to_string(), e.to_string()))?;
        let user_re = Regex::new(spec.get("user").map_or("", String::as_str))
            .map_err(|e| Error::Invalid("user regex".to_string(), e.to_string()))?;
        let comment_re = Regex::new(spec.get("comment").map_or("", String::as_str))
            .map_err(|e| Error::Invalid("comment regex".to_string(), e.to_string()))?;
        let uuid_spec = match spec.get("uuid") {
            Some(value) => Some(normalize_uuid_input(value)?),
            None => None,
        };
        let uuid_re = Regex::new(uuid_spec.as_deref().unwrap_or(""))
            .map_err(|e| Error::Invalid("uuid regex".to_string(), e.to_string()))?;
        let conn = self.inner.connect()?;
        let rows = map_turso(block_on(query_all_credentials(&conn)))?;
        let mut entries = Vec::new();
        let comment_filter = spec.get("comment").cloned();
        let filter_comment = spec.contains_key("comment");
        let filter_comment_empty = comment_filter.as_deref().is_some_and(str::is_empty);
        for (id, uuid, comment) in rows {
            if !service_re.is_match(id.service.as_str()) {
                continue;
            }
            if !user_re.is_match(id.user.as_str()) {
                continue;
            }
            if !uuid_re.is_match(uuid.as_str()) {
                continue;
            }
            if filter_comment {
                if filter_comment_empty {
                    // empty comment ("") matches only rows with no comment
                    if comment.as_deref().is_some_and(|value| !value.is_empty()) {
                        continue;
                    }
                } else {
                    // non-empty comment must match
                    match comment.as_ref() {
                        Some(text) if comment_re.is_match(text.as_str()) => {}
                        _ => continue,
                    }
                }
            }
            let credential = DbKeyCredential {
                inner: Arc::clone(&self.inner),
                id,
                uuid: Some(uuid),
                comment: None,
            };
            entries.push(Entry::new_with_credential(Arc::new(credential)));
        }
        Ok(entries)
    }

    fn as_any(&self) -> &dyn std::any::Any {
        self
    }

    fn persistence(&self) -> CredentialPersistence {
        CredentialPersistence::UntilDelete
    }

    fn debug_fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        fmt::Debug::fmt(self, f)
    }
}

impl DbKeyCredential {
    fn get_secret_zeroizing(&self) -> Result<Zeroizing<Vec<u8>>> {
        validate_service_user(&self.id.service, &self.id.user)?;
        let conn = self.inner.connect()?;
        if let Some(uuid) = &self.uuid {
            let match_result = map_turso(block_on(fetch_secret_by_key(&conn, &self.id, uuid)))?;
            match match_result {
                LookupResult::None => Err(Error::NoEntry),
                LookupResult::One(secret) => Ok(secret),
                LookupResult::Ambiguous(uuids) => Err(Error::Ambiguous(ambiguous_entries(
                    &Arc::clone(&self.inner),
                    &self.id,
                    uuids,
                ))),
            }
        } else {
            let match_result = map_turso(block_on(fetch_secret_by_id(&conn, &self.id)))?;
            match match_result {
                LookupResult::None => Err(Error::NoEntry),
                LookupResult::One(secret) => Ok(secret),
                LookupResult::Ambiguous(uuids) => Err(Error::Ambiguous(ambiguous_entries(
                    &Arc::clone(&self.inner),
                    &self.id,
                    uuids,
                ))),
            }
        }
    }

    async fn set_secret_unambiguous(
        &self,
        conn: &Connection,
        make_secret_value: &dyn Fn() -> Value,
        make_comment_value: &dyn Fn() -> Value,
    ) -> Result<()> {
        let uuid = new_uuid();
        let _ = conn.execute(
            "INSERT INTO credentials (service, user, uuid, secret, comment) VALUES (?1, ?2, ?3, ?4, ?5) \
            ON CONFLICT(service, user) DO UPDATE SET secret = excluded.secret",
            (
                self.id.service.as_str(),
                self.id.user.as_str(),
                uuid.as_str(),
                make_secret_value(),
                make_comment_value(),
            ),
        )
        .await.map_err(map_turso_err)?;
        Ok(())
    }

    async fn set_secret_with_uuid(
        &self,
        conn: &Connection,
        uuid: &str,
        make_secret_value: &dyn Fn() -> Value,
        make_comment_value: &dyn Fn() -> Value,
    ) -> Result<()> {
        let updated = conn
            .execute(
                "UPDATE credentials SET secret = ?1 WHERE service = ?2 AND user = ?3 AND uuid = ?4",
                (
                    make_secret_value(),
                    self.id.service.as_str(),
                    self.id.user.as_str(),
                    uuid,
                ),
            )
            .await
            .map_err(map_turso_err)?;
        if updated > 0 {
            return Ok(());
        }
        if !self.inner.allow_ambiguity {
            let uuids = fetch_uuids(conn, &self.id).await.map_err(map_turso_err)?;
            match uuids.len() {
                0 => {}
                1 => {
                    if uuids[0] != uuid {
                        return Err(Error::Invalid(
                            "uuid".to_string(),
                            "can't create ambiguous credential for service/user".to_string(),
                        ));
                    }
                }
                _ => {
                    // if ambiguity is not allowed, unique index should have prevented this case
                    return Err(Error::PlatformFailure(format!(
                        "Database is in an invalid state: ambiguity not allowed, but multiple entries found for {:?}",
                        self.id
                    ).into()));
                }
            }
        }
        self.insert_credential(conn, uuid, make_secret_value(), make_comment_value())
            .await?;
        Ok(())
    }

    async fn set_secret_without_uuid(
        &self,
        conn: &Connection,
        make_secret_value: &dyn Fn() -> Value,
        make_comment_value: &dyn Fn() -> Value,
    ) -> Result<()> {
        let uuids = fetch_uuids(conn, &self.id).await.map_err(map_turso_err)?;
        match uuids.len() {
            0 => {
                let uuid = new_uuid();
                self.insert_credential(
                    conn,
                    uuid.as_str(),
                    make_secret_value(),
                    make_comment_value(),
                )
                .await?;
                Ok(())
            }
            1 => {
                conn.execute(
                    "UPDATE credentials SET secret = ?1 WHERE service = ?2 AND user = ?3 AND uuid = ?4",
                    (
                        make_secret_value(),
                        self.id.service.as_str(),
                        self.id.user.as_str(),
                        uuids[0].as_str(),
                    ),
                )
                .await
                .map_err(map_turso_err)?;
                Ok(())
            }
            _ => Err(Error::Ambiguous(ambiguous_entries(
                &self.inner,
                &self.id,
                uuids,
            ))),
        }
    }

    async fn set_secret_in_tx(
        &self,
        conn: &Connection,
        make_secret_value: &dyn Fn() -> Value,
        make_comment_value: &dyn Fn() -> Value,
    ) -> Result<()> {
        if let Some(uuid) = &self.uuid {
            self.set_secret_with_uuid(conn, uuid.as_str(), make_secret_value, make_comment_value)
                .await
        } else {
            self.set_secret_without_uuid(conn, make_secret_value, make_comment_value)
                .await
        }
    }

    async fn finish_tx(conn: &Connection, result: Result<()>) -> Result<()> {
        match result {
            Ok(()) => {
                conn.execute("COMMIT", ()).await.map_err(map_turso_err)?;
                Ok(())
            }
            Err(err) => {
                if let Err(e2) = conn.execute("ROLLBACK", ()).await {
                    log::error!(
                        "While handling set_secret error ({err:?}). attempted ROLLBACK, which encountered secondary error: {e2:?}"
                    );
                }
                Err(err)
            }
        }
    }
}

impl CredentialApi for DbKeyCredential {
    fn set_secret(&self, secret: &[u8]) -> Result<()> {
        validate_service_user(&self.id.service, &self.id.user)?;
        validate_secret(secret)?;
        let make_secret_value = || Value::Blob(secret.to_vec());
        let make_comment_value = || comment_value(self.comment.as_ref());
        let conn = self.inner.connect()?;
        if self.uuid.is_none() && !self.inner.allow_ambiguity {
            return block_on(self.set_secret_unambiguous(
                &conn,
                &make_secret_value,
                &make_comment_value,
            ));
        }
        block_on(async {
            conn.execute("BEGIN IMMEDIATE", ())
                .await
                .map_err(map_turso_err)?;
            let result = self
                .set_secret_in_tx(&conn, &make_secret_value, &make_comment_value)
                .await;
            Self::finish_tx(&conn, result).await
        })
    }

    fn get_secret(&self) -> Result<Vec<u8>> {
        let secret = self.get_secret_zeroizing()?;
        Ok(take_zeroizing_vec(secret))
    }

    fn get_attributes(&self) -> Result<HashMap<String, String>> {
        validate_service_user(&self.id.service, &self.id.user)?;
        let conn = self.inner.connect()?;
        if let Some(uuid) = &self.uuid {
            let match_result = map_turso(block_on(fetch_comment_by_key(&conn, &self.id, uuid)))?;
            match match_result {
                LookupResult::None => Err(Error::NoEntry),
                LookupResult::One(comment) => Ok(attributes_for_uuid(uuid.as_str(), comment)),
                LookupResult::Ambiguous(uuids) => Err(Error::Ambiguous(ambiguous_entries(
                    &self.inner,
                    &self.id,
                    uuids,
                ))),
            }
        } else {
            let match_result = map_turso(block_on(fetch_comment_by_id(&conn, &self.id)))?;
            match match_result {
                LookupResult::None => Err(Error::NoEntry),
                LookupResult::One(row) => Ok(attributes_for_uuid(row.uuid.as_str(), row.comment)),
                LookupResult::Ambiguous(uuids) => Err(Error::Ambiguous(ambiguous_entries(
                    &self.inner,
                    &self.id,
                    uuids,
                ))),
            }
        }
    }

    fn update_attributes(&self, attrs: &HashMap<&str, &str>) -> Result<()> {
        parse_attributes(&["comment"], Some(attrs))?;
        let comment = attrs.get("comment").map(ToString::to_string);
        let has_comment = attrs.contains_key("comment");
        if !has_comment {
            self.get_attributes()?;
            return Ok(());
        }
        let comment = comment.filter(|value| !value.is_empty());
        let make_comment_value = || comment_value(comment.as_ref());
        let conn = self.inner.connect()?;
        block_on(async {
            conn.execute("BEGIN IMMEDIATE", ())
                .await
                .map_err(map_turso_err)?;
            let result = match &self.uuid {
                Some(uuid) => {
                    let uuids = fetch_uuids_by_key(&conn, &self.id, uuid)
                        .await
                        .map_err(map_turso_err)?;
                    match uuids.len() {
                        0 => Err(Error::NoEntry),
                        1 => {
                            conn.execute(
                                "UPDATE credentials SET comment = ?1 WHERE service = ?2 AND user = ?3 AND uuid = ?4",
                                (
                                    make_comment_value(),
                                    self.id.service.as_str(),
                                    self.id.user.as_str(),
                                    uuid.as_str(),
                                ),
                            )
                            .await
                            .map_err(map_turso_err)?;
                            Ok(())
                        }
                        _ => Err(Error::Ambiguous(ambiguous_entries(
                            &self.inner,
                            &self.id,
                            uuids,
                        ))),
                    }
                }
                None if self.inner.allow_ambiguity => {
                    let uuids = fetch_uuids(&conn, &self.id).await.map_err(map_turso_err)?;
                    match uuids.len() {
                        0 => Err(Error::NoEntry),
                        1 => {
                            conn.execute(
                                "UPDATE credentials SET comment = ?1 WHERE service = ?2 AND user = ?3 AND uuid = ?4",
                                (
                                    make_comment_value(),
                                    self.id.service.as_str(),
                                    self.id.user.as_str(),
                                    uuids[0].as_str(),
                                ),
                            )
                            .await
                            .map_err(map_turso_err)?;
                            Ok(())
                        }
                        _ => Err(Error::Ambiguous(ambiguous_entries(
                            &self.inner,
                            &self.id,
                            uuids,
                        ))),
                    }
                }
                None => {
                    let updated = conn
                        .execute(
                            "UPDATE credentials SET comment = ?1 WHERE service = ?2 AND user = ?3",
                            (
                                make_comment_value(),
                                self.id.service.as_str(),
                                self.id.user.as_str(),
                            ),
                        )
                        .await
                        .map_err(map_turso_err)?;
                    if updated == 0 {
                        Err(Error::NoEntry)
                    } else {
                        Ok(())
                    }
                }
            };
            match result {
                Ok(()) => {
                    conn.execute("COMMIT", ()).await.map_err(map_turso_err)?;
                    Ok(())
                }
                Err(err) => {
                    // attempt rollback but if rollback fails report original error
                    let _ = conn.execute("ROLLBACK", ()).await;
                    Err(err)
                }
            }
        })
    }

    fn delete_credential(&self) -> Result<()> {
        validate_service_user(&self.id.service, &self.id.user)?;
        let conn = self.inner.connect()?;
        if let Some(uuid) = &self.uuid {
            let deleted = map_turso(block_on(conn.execute(
                "DELETE FROM credentials WHERE service = ?1 AND user = ?2 AND uuid = ?3",
                (
                    self.id.service.as_str(),
                    self.id.user.as_str(),
                    uuid.as_str(),
                ),
            )))?;
            if deleted == 0 {
                Err(Error::NoEntry)
            } else {
                Ok(())
            }
        } else {
            let uuids = map_turso(block_on(fetch_uuids(&conn, &self.id)))?;
            match uuids.len() {
                0 => Err(Error::NoEntry),
                1 => {
                    map_turso(block_on(conn.execute(
                        "DELETE FROM credentials WHERE service = ?1 AND user = ?2 AND uuid = ?3",
                        (
                            self.id.service.as_str(),
                            self.id.user.as_str(),
                            uuids[0].as_str(),
                        ),
                    )))?;
                    Ok(())
                }
                _ => Err(Error::Ambiguous(ambiguous_entries(
                    &self.inner,
                    &self.id,
                    uuids,
                ))),
            }
        }
    }

    fn get_credential(&self) -> Result<Option<Arc<Credential>>> {
        validate_service_user(&self.id.service, &self.id.user)?;
        let conn = self.inner.connect()?;
        if let Some(uuid) = &self.uuid {
            let uuids = map_turso(block_on(fetch_uuids_by_key(&conn, &self.id, uuid)))?;
            match uuids.len() {
                0 => Err(Error::NoEntry),
                1 => Ok(Some(Arc::new(DbKeyCredential {
                    inner: Arc::clone(&self.inner),
                    id: self.id.clone(),
                    uuid: Some(uuid.clone()),
                    comment: None,
                }))),
                _ => Err(Error::Ambiguous(ambiguous_entries(
                    &self.inner,
                    &self.id,
                    uuids,
                ))),
            }
        } else {
            let uuids = map_turso(block_on(fetch_uuids(&conn, &self.id)))?;
            match uuids.len() {
                0 => Err(Error::NoEntry),
                1 => Ok(Some(Arc::new(DbKeyCredential {
                    inner: Arc::clone(&self.inner),
                    id: self.id.clone(),
                    uuid: Some(uuids[0].clone()),
                    comment: None,
                }))),
                _ => Err(Error::Ambiguous(ambiguous_entries(
                    &self.inner,
                    &self.id,
                    uuids,
                ))),
            }
        }
    }

    fn get_specifiers(&self) -> Option<(String, String)> {
        Some((self.id.service.clone(), self.id.user.clone()))
    }

    fn as_any(&self) -> &dyn std::any::Any {
        self
    }

    fn debug_fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        fmt::Debug::fmt(self, f)
    }
}

fn init_schema(conn: &Connection, allow_ambiguity: bool, index_always: bool) -> Result<()> {
    map_turso(block_on(conn.execute(
        "CREATE TABLE IF NOT EXISTS credentials (service TEXT NOT NULL, user TEXT NOT NULL, uuid TEXT NOT NULL, secret BLOB NOT NULL, comment TEXT)",
        (),
    )))?;
    map_turso(block_on(conn.execute(
        "CREATE TABLE IF NOT EXISTS keystore_meta (key TEXT NOT NULL PRIMARY KEY, value TEXT NOT NULL)",
        (),
    )))?;
    ensure_schema_version(conn)?;
    if !allow_ambiguity {
        // unique index used to help ensure non-ambiguity of (service,user)
        map_turso(block_on(conn.execute(
            "CREATE UNIQUE INDEX IF NOT EXISTS uidx_credentials_service_user ON credentials (service, user)",
            (),
        )))?;
    } else if index_always {
        // Performance tradeoffs: this index roughly doubles the file size.
        // - For keystores with ~100 entries, it saves 0.1ms per lookup (.17 vs .28 ms).
        // - For ~1000 entries, the index saves ~1ms per lookup (0.1 vs 1.4 ms)
        // - Measured on a m3 macbook air.
        map_turso(block_on(conn.execute(
             "CREATE INDEX IF NOT EXISTS idx_credentials_service_user ON credentials (service, user)",
             (),
            )))?;
    }
    Ok(())
}

fn ensure_schema_version(conn: &Connection) -> Result<()> {
    map_turso(block_on(async {
        let mut rows = conn
            .query(
                "SELECT value FROM keystore_meta WHERE key = 'schema_version'",
                (),
            )
            .await?;
        if let Some(row) = rows.next().await? {
            let value = value_to_string(row.get_value(0)?, "schema_version")?;
            let version = value.parse::<u32>().map_err(|_| {
                turso::Error::ConversionFailure(format!("invalid schema_version value: {value}"))
            })?;
            if version != SCHEMA_VERSION {
                return Err(turso::Error::ConversionFailure(format!(
                    "unsupported schema version: {version}"
                )));
            }
        } else {
            conn.execute(
                "INSERT INTO keystore_meta (key, value) VALUES ('schema_version', ?1)",
                (SCHEMA_VERSION.to_string(),),
            )
            .await?;
        }
        Ok(())
    }))
}

async fn query_all_credentials(
    conn: &Connection,
) -> turso::Result<Vec<(CredId, String, Option<String>)>> {
    let mut rows = conn
        .query("SELECT service, user, uuid, comment FROM credentials", ())
        .await?;
    let mut results = Vec::new();
    while let Some(row) = rows.next().await? {
        let service = value_to_string(row.get_value(0)?, "service")?;
        let user = value_to_string(row.get_value(1)?, "user")?;
        let uuid = value_to_string(row.get_value(2)?, "uuid")?;
        let comment = value_to_option_string(row.get_value(3)?, "comment")?;
        results.push((CredId { service, user }, uuid, comment));
    }
    Ok(results)
}

/// Returns true if the `credentials` schema enforces `(service, user)`
/// uniqueness, either via a unique index or a table-level unique constraint.
async fn schema_has_unique_service_user(conn: &Connection) -> turso::Result<bool> {
    let mut rows = conn
        .query(
            "SELECT sql FROM sqlite_master \
             WHERE (type = 'index' AND tbl_name = 'credentials') \
                OR (type = 'table' AND name = 'credentials') \
             AND sql IS NOT NULL",
            (),
        )
        .await?;
    while let Some(row) = rows.next().await? {
        match row.get_value(0)? {
            Value::Text(sql) if is_unique_service_user_sql(sql.as_str()) => return Ok(true),
            _ => {}
        }
    }
    Ok(false)
}

/// True if a `sqlite_master.sql` definition enforces uniqueness on
/// `(service, user)` (whitespace/quote/case insensitive).
fn is_unique_service_user_sql(sql: &str) -> bool {
    let normalized: String = sql
        .chars()
        .filter(|c| !c.is_whitespace() && *c != '"' && *c != '`')
        .flat_map(char::to_lowercase)
        .collect();
    normalized.contains("unique") && normalized.contains("(service,user)")
}

async fn fetch_uuids(conn: &Connection, id: &CredId) -> turso::Result<Vec<String>> {
    let mut rows = conn
        .query(
            "SELECT uuid FROM credentials WHERE service = ?1 AND user = ?2",
            (id.service.as_str(), id.user.as_str()),
        )
        .await?;
    let mut uuids = Vec::new();
    while let Some(row) = rows.next().await? {
        let uuid = value_to_string(row.get_value(0)?, "uuid")?;
        uuids.push(uuid);
    }
    Ok(uuids)
}

async fn fetch_secret_by_key(
    conn: &Connection,
    id: &CredId,
    uuid: &str,
) -> turso::Result<LookupResult<Zeroizing<Vec<u8>>>> {
    let mut rows = conn
        .query(
            "SELECT secret FROM credentials WHERE service = ?1 AND user = ?2 AND uuid = ?3",
            (id.service.as_str(), id.user.as_str(), uuid),
        )
        .await?;
    let mut secrets = Vec::new();
    while let Some(row) = rows.next().await? {
        let secret = value_to_secret(row.get_value(0)?, "secret")?;
        secrets.push(secret);
    }
    match secrets.len() {
        0 => Ok(LookupResult::None),
        1 => Ok(LookupResult::One(
            secrets.into_iter().next().expect("secret for single match"),
        )),
        _ => Ok(LookupResult::Ambiguous(vec![
            uuid.to_string();
            secrets.len()
        ])),
    }
}

async fn fetch_comment_by_key(
    conn: &Connection,
    id: &CredId,
    uuid: &str,
) -> turso::Result<LookupResult<Option<String>>> {
    let mut rows = conn
        .query(
            "SELECT comment FROM credentials WHERE service = ?1 AND user = ?2 AND uuid = ?3",
            (id.service.as_str(), id.user.as_str(), uuid),
        )
        .await?;
    let mut comments = Vec::new();
    while let Some(row) = rows.next().await? {
        let comment = value_to_option_string(row.get_value(0)?, "comment")?;
        comments.push(comment);
    }
    match comments.len() {
        0 => Ok(LookupResult::None),
        1 => Ok(LookupResult::One(
            comments
                .into_iter()
                .next()
                .expect("comment for single match"),
        )),
        _ => Ok(LookupResult::Ambiguous(vec![
            uuid.to_string();
            comments.len()
        ])),
    }
}

async fn fetch_secret_by_id(
    conn: &Connection,
    id: &CredId,
) -> turso::Result<LookupResult<Zeroizing<Vec<u8>>>> {
    let uuids = fetch_uuids(conn, id).await?;
    match uuids.len() {
        0 => Ok(LookupResult::None),
        1 => fetch_secret_by_key(conn, id, uuids[0].as_str()).await,
        _ => Ok(LookupResult::Ambiguous(uuids)),
    }
}

async fn fetch_comment_by_id(
    conn: &Connection,
    id: &CredId,
) -> turso::Result<LookupResult<CommentRow>> {
    let uuids = fetch_uuids(conn, id).await?;
    match uuids.len() {
        0 => Ok(LookupResult::None),
        1 => {
            let uuid = uuids.into_iter().next().expect("uuid");
            match fetch_comment_by_key(conn, id, uuid.as_str()).await? {
                LookupResult::None => Ok(LookupResult::None),
                LookupResult::One(comment) => Ok(LookupResult::One(CommentRow { uuid, comment })),
                LookupResult::Ambiguous(uuids) => Ok(LookupResult::Ambiguous(uuids)),
            }
        }
        _ => Ok(LookupResult::Ambiguous(uuids)),
    }
}

async fn fetch_uuids_by_key(
    conn: &Connection,
    id: &CredId,
    uuid: &str,
) -> turso::Result<Vec<String>> {
    let mut rows = conn
        .query(
            "SELECT uuid FROM credentials WHERE service = ?1 AND user = ?2 AND uuid = ?3",
            (id.service.as_str(), id.user.as_str(), uuid),
        )
        .await?;
    let mut uuids = Vec::new();
    while let Some(row) = rows.next().await? {
        let uuid = value_to_string(row.get_value(0)?, "uuid")?;
        uuids.push(uuid);
    }
    Ok(uuids)
}

fn ambiguous_entries(inner: &Arc<DbKeyStoreInner>, id: &CredId, uuids: Vec<String>) -> Vec<Entry> {
    uuids
        .into_iter()
        .map(|uuid| {
            Entry::new_with_credential(Arc::new(DbKeyCredential {
                inner: Arc::clone(inner),
                id: id.clone(),
                uuid: Some(uuid),
                comment: None,
            }))
        })
        .collect()
}

fn attributes_for_uuid(uuid: &str, comment: Option<String>) -> HashMap<String, String> {
    let mut attrs = HashMap::new();
    attrs.insert("uuid".to_string(), uuid.to_string());
    if let Some(comment) = comment {
        attrs.insert("comment".to_string(), comment);
    }
    attrs
}

fn comment_value(comment: Option<&String>) -> Value {
    match comment {
        Some(value) if !value.is_empty() => Value::Text(value.clone()),
        _ => Value::Null,
    }
}

fn normalize_uuid_input(value: &str) -> Result<String> {
    let lower = value.to_ascii_lowercase();
    let uuid = uuid::Uuid::try_parse(&lower)
        .map_err(|_| Error::Invalid("uuid".to_string(), "invalid uuid format".to_string()))?;
    if uuid.to_string() != lower {
        return Err(Error::Invalid(
            "uuid".to_string(),
            "invalid uuid format".to_string(),
        ));
    }
    Ok(lower)
}

fn take_zeroizing_vec(mut value: Zeroizing<Vec<u8>>) -> Vec<u8> {
    std::mem::take(&mut *value)
}

// Database configuration
// - Enable Write-Ahead Logging for better concurrency (less blocking)
// - Sets busy timeout - how long to wait when db is locked by another connection before returning error
fn configure_connection(conn: &Connection) -> Result<()> {
    map_turso(block_on(async {
        let mut rows = conn.query("PRAGMA journal_mode=WAL", ()).await?;
        let _ = rows.next().await?;
        let busy_stmt = format!("PRAGMA busy_timeout = {BUSY_TIMEOUT_MS}");
        conn.execute(busy_stmt.as_str(), ()).await?;
        Ok(())
    }))
}

/// Opens database. Retries with exponential backoff if the file is locked.
fn open_db_with_retry(
    path_str: &str,
    encryption_opts: Option<&EncryptionOpts>,
    vfs: Option<&str>,
) -> Result<Database> {
    let mut retries = OPEN_LOCK_RETRIES;
    let mut backoff_ms = OPEN_LOCK_BACKOFF_MS;
    loop {
        let mut builder = Builder::new_local(path_str);
        if let Some(opts) = encryption_opts {
            // key stays zeroizing on our side; see turso_encryption_opts for the boundary note
            builder = builder
                .experimental_encryption(true)
                .with_encryption(turso_encryption_opts(opts));
        }
        if let Some(vfs) = vfs {
            builder = builder.with_io(vfs.to_string());
        }
        match block_on(builder.build()) {
            Ok(db) => return Ok(db),
            Err(err) => {
                check_decryption_error(&err)?;
                if retries == 0 || !is_turso_locking_error(&err) {
                    return Err(map_turso_err(err));
                }
                retries -= 1;
                let nanos = SystemTime::now()
                    .duration_since(UNIX_EPOCH)
                    .unwrap_or_default()
                    .subsec_nanos();
                let jitter = u64::from(nanos % 20);
                std::thread::sleep(std::time::Duration::from_millis(backoff_ms + jitter));
                backoff_ms = (backoff_ms * 2).min(OPEN_LOCK_BACKOFF_MAX_MS);
            }
        }
    }
}

fn retry_turso_locking<T>(mut op: impl FnMut() -> turso::Result<T>) -> Result<T> {
    let mut retries = OPEN_LOCK_RETRIES;
    let mut backoff_ms = OPEN_LOCK_BACKOFF_MS;
    loop {
        match op() {
            Ok(value) => return Ok(value),
            Err(err) => {
                if retries == 0 || !is_turso_locking_error(&err) {
                    return Err(map_turso_err(err));
                }
                retries -= 1;
                let nanos = SystemTime::now()
                    .duration_since(UNIX_EPOCH)
                    .unwrap_or_default()
                    .subsec_nanos();
                let jitter = u64::from(nanos % 20);
                std::thread::sleep(std::time::Duration::from_millis(backoff_ms + jitter));
                backoff_ms = (backoff_ms * 2).min(OPEN_LOCK_BACKOFF_MAX_MS);
            }
        }
    }
}

fn is_turso_locking_error(err: &turso::Error) -> bool {
    let text = err.to_string().to_lowercase();
    text.contains("locking error")
        || text.contains("file is locked")
        || text.contains("database is locked")
        || text.contains("database is busy")
        || text.contains("sqlite_busy")
        || text.contains("sqlite_locked")
}

fn check_decryption_error(err: &turso::Error) -> Result<()> {
    let text = err.to_string();
    if text.starts_with("Decryption failed") {
        return Err(keyring_core::Error::NoStorageAccess(Box::new(
            turso::Error::Error(format!("Invalid encryption key or cipher. {text}")),
        )));
    }
    Ok(())
}

fn value_to_string(value: Value, field: &str) -> turso::Result<String> {
    match value {
        Value::Text(text) => Ok(text),
        Value::Blob(blob) => String::from_utf8(blob)
            .map_err(|e| turso::Error::ConversionFailure(format!("invalid utf8 for {field}: {e}"))),
        other => Err(turso::Error::ConversionFailure(format!(
            "unexpected value for {field}: {other:?}"
        ))),
    }
}

fn value_to_secret(value: Value, field: &str) -> turso::Result<Zeroizing<Vec<u8>>> {
    match value {
        Value::Blob(blob) => Ok(Zeroizing::new(blob)),
        Value::Text(text) => Ok(Zeroizing::new(text.into_bytes())),
        other => Err(turso::Error::ConversionFailure(format!(
            "unexpected value for {field}: {other:?}"
        ))),
    }
}

fn value_to_option_string(value: Value, field: &str) -> turso::Result<Option<String>> {
    match value {
        Value::Null => Ok(None),
        Value::Text(text) => Ok(Some(text)),
        Value::Blob(blob) => String::from_utf8(blob)
            .map(Some)
            .map_err(|e| turso::Error::ConversionFailure(format!("invalid utf8 for {field}: {e}"))),
        other => Err(turso::Error::ConversionFailure(format!(
            "unexpected value for {field}: {other:?}"
        ))),
    }
}

fn ensure_parent_dir(path: &Path) -> Result<()> {
    let parent = path
        .parent()
        .ok_or_else(|| Error::Invalid("path".to_string(), "path has no parent".to_string()))?;
    if parent.as_os_str().is_empty() {
        return Ok(());
    }
    std::fs::create_dir_all(parent).map_err(|e| Error::PlatformFailure(Box::new(e)))
}

/// confirm service and user are non-empty and within length bounds
fn validate_service_user(service: &str, user: &str) -> Result<()> {
    if service.is_empty() {
        return Err(Error::Invalid(
            "service".to_string(),
            "service is empty".to_string(),
        ));
    }
    if user.is_empty() {
        return Err(Error::Invalid(
            "user".to_string(),
            "user is empty".to_string(),
        ));
    }
    if service.len() > MAX_NAME_LEN as usize {
        return Err(Error::TooLong("service".to_string(), MAX_NAME_LEN));
    }
    if user.len() > MAX_NAME_LEN as usize {
        return Err(Error::TooLong("user".to_string(), MAX_NAME_LEN));
    }
    Ok(())
}

/// confirm secret is within length bounds
fn validate_secret(secret: &[u8]) -> Result<()> {
    validate_secret_len(secret.len())
}

/// confirm secret length is within bounds (without needing the bytes)
fn validate_secret_len(len: usize) -> Result<()> {
    if len > MAX_SECRET_LEN as usize {
        return Err(Error::TooLong("secret".to_string(), MAX_SECRET_LEN));
    }
    Ok(())
}

fn map_turso<T>(result: std::result::Result<T, turso::Error>) -> Result<T> {
    result.map_err(map_turso_err)
}

fn map_turso_err(err: turso::Error) -> Error {
    Error::PlatformFailure(Box::new(err))
}

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

    fn new_store(path: &Path) -> Arc<DbKeyStore> {
        let config = DbKeyStoreConfig {
            path: path.to_path_buf(),
            ..Default::default()
        };
        DbKeyStore::new(config).expect("failed to create store")
    }

    fn build_entry(store: &DbKeyStore, service: &str, user: &str) -> Entry {
        store
            .build(service, user, None)
            .expect("failed to build entry")
    }

    fn set_password(entry: &Entry, password: &str) -> Result<()> {
        entry.set_password(password)
    }

    fn set_secret(entry: &Entry, secret: &[u8]) -> Result<()> {
        entry.set_secret(secret)
    }

    fn get_password(entry: &Entry) -> Result<Zeroizing<String>> {
        Ok(Zeroizing::new(entry.get_password()?))
    }

    // test that non-existent parent dir is created on db open
    #[test]
    fn create_store_creates_parent_dir() {
        let dir = tempfile::tempdir().expect("tempdir");
        let db_path = dir.path().join("nested").join("deeply").join("keystore.db");
        let parent = db_path.parent().expect("parent");
        assert!(!parent.exists());

        let config = DbKeyStoreConfig {
            path: db_path.clone(),
            ..Default::default()
        };
        let store = DbKeyStore::new(config).expect("create store");
        assert!(parent.is_dir());

        let entry = build_entry(&store, "demo", "alice");
        set_password(&entry, "dromomeryx").expect("set_password");
    }

    // test round-trip set and search
    #[test]
    fn set_password_then_search_finds_password() {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("keystore.db");
        let store = new_store(&path);
        let entry = build_entry(&store, "demo", "alice");
        set_password(&entry, "dromomeryx").expect("set_password");

        let mut spec = HashMap::new();
        spec.insert("service", "demo");
        spec.insert("user", "alice");
        let results = store.search(&spec).expect("search");
        assert_eq!(results.len(), 1);
        let password = get_password(&results[0]).expect("get_password");
        assert_eq!(password.as_str(), "dromomeryx");
    }

    // test with comment search
    #[test]
    fn comment_attributes_round_trip() {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("keystore.db");
        let store = new_store(&path);
        let entry = build_entry(&store, "demo", "alice");
        set_password(&entry, "dromomeryx").expect("set_password");

        let update = HashMap::from([("comment", "note")]);
        entry.update_attributes(&update).expect("update_attributes");
        let attrs = entry.get_attributes().expect("get_attributes");
        assert_eq!(attrs.get("comment"), Some(&"note".to_string()));
        assert!(attrs.contains_key("uuid"));

        let mut spec = HashMap::new();
        spec.insert("service", "demo");
        spec.insert("user", "alice");
        spec.insert("comment", "note");
        let results = store.search(&spec).expect("search");
        assert_eq!(results.len(), 1);

        let uuid = attrs.get("uuid").cloned().expect("get uuid");
        let mut spec = HashMap::new();
        spec.insert("service", "demo");
        spec.insert("user", "alice");
        spec.insert("uuid", uuid.as_str());
        let results = store.search(&spec).expect("search");
        assert_eq!(results.len(), 1);
    }

    #[test]
    fn comment_with_password_round_trip() {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("keystore.db");
        let store = new_store(&path);
        let entry = build_entry(&store, "demo", "alice");
        set_password(&entry, "dromomeryx").expect("set_password");

        // set a comment attribute
        let update = HashMap::from([("comment", "note")]);
        entry.update_attributes(&update).expect("update_attributes");

        // then search by comment
        let mut spec = HashMap::new();
        spec.insert("service", "demo");
        spec.insert("user", "alice");
        spec.insert("comment", "note");
        let results = store.search(&spec).expect("search");
        assert_eq!(results.len(), 1);

        let found = &results[0];
        let password = get_password(found).expect("password with comment");
        assert_eq!(password.as_str(), "dromomeryx");
        let attrs = found.get_attributes().expect("get_attributes");
        assert_eq!(attrs.get("comment"), Some(&"note".to_string()));
        assert!(attrs.contains_key("uuid"));
    }

    #[test]
    fn build_with_comment_modifier_sets_comment() -> Result<()> {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("keystore.db");
        let store = new_store(&path);
        let entry = store.build(
            "demo",
            "alice",
            Some(&HashMap::from([("comment", "initial")])),
        )?;
        set_password(&entry, "dromomeryx")?;

        let attrs = entry.get_attributes()?;
        assert_eq!(attrs.get("comment"), Some(&"initial".to_string()));
        Ok(())
    }

    #[test]
    fn in_memory_store_round_trip() -> Result<()> {
        let config = DbKeyStoreConfig {
            vfs: Some("memory".to_string()),
            ..Default::default()
        };
        let store = DbKeyStore::new(config)?;
        let entry = build_entry(&store, "demo", "alice");
        set_password(&entry, "dromomeryx")?;

        let results = store.search(&HashMap::from([("service", "demo"), ("user", "alice")]))?;
        assert_eq!(results.len(), 1);
        let password = get_password(&results[0])?;
        assert_eq!(password.as_str(), "dromomeryx");
        Ok(())
    }

    // test that unique users in same service have unique keys
    #[test]
    fn stores_separate_service_user_pairs() -> Result<()> {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("keystore.db");
        let store = new_store(&path);

        let entry = build_entry(&store, "myapp", "user1");
        set_password(&entry, "pw1")?;
        let entry = build_entry(&store, "myapp", "user2");
        set_password(&entry, "pw2")?;
        let entry = build_entry(&store, "myapp", "user3");
        set_password(&entry, "pw3")?;

        let results = store.search(&HashMap::from([("service", "myapp"), ("user", "user1")]))?;
        assert_eq!(results.len(), 1);
        let password = get_password(&results[0])?;
        assert_eq!(password.as_str(), "pw1");

        let results = store.search(&HashMap::from([("service", "myapp"), ("user", "user2")]))?;
        assert_eq!(results.len(), 1);
        let password = get_password(&results[0])?;
        assert_eq!(password.as_str(), "pw2");

        let results = store.search(&HashMap::from([("service", "myapp"), ("user", "user3")]))?;
        assert_eq!(results.len(), 1);
        let password = get_password(&results[0])?;
        assert_eq!(password.as_str(), "pw3");
        Ok(())
    }

    // search with regex
    #[test]
    fn search_regex() -> Result<()> {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("keystore.db");
        let store = new_store(&path);

        let entry = build_entry(&store, "myapp", "user1");
        set_password(&entry, "pw1")?;
        let entry = build_entry(&store, "myapp", "user2");
        set_password(&entry, "pw2")?;
        let entry = build_entry(&store, "myapp", "user3");
        set_password(&entry, "pw3")?;
        let entry = build_entry(&store, "other-app", "user1");
        set_password(&entry, "pw4")?;

        // regex search: all apps, user1
        let results = store.search(&HashMap::from([("service", ".*app"), ("user", "user1")]))?;
        assert_eq!(results.len(), 2, "search *app, user1");

        // regex search _or_
        let results = store.search(&HashMap::from([
            ("service", "myapp"),
            ("user", "user1|user2"),
        ]))?;
        assert_eq!(results.len(), 2, "search regex OR");

        Ok(())
    }

    // search with partial hashmap
    #[test]
    fn search_partial() -> Result<()> {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("keystore.db");
        let store = new_store(&path);

        // empty db has no entries
        let results = store.search(&HashMap::new())?;
        assert_eq!(results.len(), 0, "empty db, no results");

        let entry = build_entry(&store, "myapp", "user1");
        set_password(&entry, "pw1")?;
        let entry = build_entry(&store, "other-app", "user1");
        set_password(&entry, "pw2")?;

        // empty search terms match all
        let results = store.search(&HashMap::new())?;
        assert_eq!(results.len(), 2, "search, empty hashmap");

        // app-only match
        let results = store.search(&HashMap::from([("service", "myapp")]))?;
        assert_eq!(results.len(), 1, "search myapp");

        // user-only match
        let results = store.search(&HashMap::from([("user", "user1")]))?;
        assert_eq!(results.len(), 2, "search user1");
        Ok(())
    }

    // replacement
    #[test]
    fn repeated_set_replaces_secret() {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("keystore.db");
        let store = new_store(&path);
        let entry = build_entry(&store, "demo", "alice");
        set_password(&entry, "first").expect("password set 1");
        set_secret(&entry, b"second").expect("password set 2");

        let mut spec = HashMap::new();
        spec.insert("service", "demo");
        spec.insert("user", "alice");
        let results = store.search(&spec).expect("search");
        assert_eq!(results.len(), 1);
        let password = get_password(&results[0]).expect("get first password");
        assert_eq!(
            password.as_str(),
            "second",
            "second password overwrites first"
        );
    }

    #[test]
    fn same_service_user_entries_share_credential() -> Result<()> {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("keystore.db");
        let store = new_store(&path);
        let entry1 = build_entry(&store, "demo", "alice");
        let entry2 = build_entry(&store, "demo", "alice");

        set_password(&entry1, "first")?;
        let password = get_password(&entry2)?;
        assert_eq!(password.as_str(), "first");

        set_password(&entry2, "second")?;
        let password = get_password(&entry1)?;
        assert_eq!(password.as_str(), "second");
        Ok(())
    }

    // deletion returns NoEntry if there is no matching entry
    #[test]
    fn remove_returns_no_entry() {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("keystore.db");
        let store = new_store(&path);
        let entry = build_entry(&store, "demo", "alice");
        set_password(&entry, "dromomeryx").expect("set password");
        entry.delete_credential().expect("delete credential");
        let err = entry.delete_credential().unwrap_err();
        assert!(matches!(err, Error::NoEntry));
    }

    // deletion actually deletes
    #[test]
    fn remove_clears_secret() {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("keystore.db");
        let store = new_store(&path);
        let entry = build_entry(&store, "service", "user");
        set_password(&entry, "dromomeryx").expect("set password");
        entry.delete_credential().expect("delete credential");

        let mut spec = HashMap::new();
        spec.insert("service", "demo");
        spec.insert("user", "alice");
        let results = store.search(&spec).expect("search");
        assert!(results.is_empty());
    }

    #[test]
    fn allow_ambiguity_allows_multiple_entries_per_user() -> Result<()> {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("keystore.db");
        let config = DbKeyStoreConfig {
            path: path.clone(),
            allow_ambiguity: true,
            ..Default::default()
        };
        let store = DbKeyStore::new(config)?;
        let uuid1 = new_uuid();
        let uuid2 = new_uuid();
        let entry1 = store.build(
            "demo",
            "alice",
            Some(&HashMap::from([
                ("uuid", uuid1.as_str()),
                ("comment", "one"),
            ])),
        )?;
        let entry2 = store.build(
            "demo",
            "alice",
            Some(&HashMap::from([
                ("uuid", uuid2.as_str()),
                ("comment", "two"),
            ])),
        )?;
        set_password(&entry1, "first")?;
        set_password(&entry2, "second")?;

        let results = store.search(&HashMap::from([("service", "demo"), ("user", "alice")]))?;
        assert_eq!(results.len(), 2);

        let entry3 = build_entry(&store, "demo", "alice");
        let err = entry3.get_password().unwrap_err();
        assert!(matches!(err, Error::Ambiguous(_)));
        Ok(())
    }

    #[test]
    fn duplicate_uuid_across_service_user_is_scoped() -> Result<()> {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("keystore.db");
        let config = DbKeyStoreConfig {
            path: path.clone(),
            allow_ambiguity: true,
            ..Default::default()
        };
        let store = DbKeyStore::new(config)?;
        let uuid = "f81d4fae-7dec-11d0-a765-00a0c91e6bf6";
        let entry1 = store.build(
            "service-a",
            "user-a",
            Some(&HashMap::from([("uuid", uuid)])),
        )?;
        let entry2 = store.build(
            "service-b",
            "user-b",
            Some(&HashMap::from([("uuid", uuid)])),
        )?;
        set_password(&entry1, "pw1")?;
        set_password(&entry2, "pw2")?;

        entry1.update_attributes(&HashMap::from([("comment", "note1")]))?;
        let attrs1 = entry1.get_attributes()?;
        assert_eq!(attrs1.get("comment"), Some(&"note1".to_string()));

        let attrs2 = entry2.get_attributes()?;
        assert!(!attrs2.contains_key("comment"));

        entry1.delete_credential()?;
        let pw2 = get_password(&entry2)?;
        assert_eq!(pw2.as_str(), "pw2");
        Ok(())
    }

    #[test]
    fn disallow_ambiguity_rejects_duplicate_uuid_entries() -> Result<()> {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("keystore.db");
        let store = new_store(&path);
        let uuid1 = new_uuid();
        let uuid2 = new_uuid();
        let entry1 = store.build(
            "demo",
            "alice",
            Some(&HashMap::from([("uuid", uuid1.as_str())])),
        )?;
        let entry2 = store.build(
            "demo",
            "alice",
            Some(&HashMap::from([("uuid", uuid2.as_str())])),
        )?;

        set_password(&entry1, "first")?;
        let err = set_password(&entry2, "second").unwrap_err();
        assert!(matches!(err, Error::Invalid(key, _) if key == "uuid"));
        Ok(())
    }

    #[test]
    fn impl_debug() -> Result<()> {
        let dir = tempfile::tempdir().expect("tempdir");

        let path = dir.path().join("keystore1.db");
        let store = new_store(&path);
        eprintln!("basic: {store:?}");

        let path = dir.path().join("keystore2.db");
        let config = DbKeyStoreConfig {
            path: path.clone(),
            encryption_opts: Some(EncryptionOpts::new(
                "aes256gcm",
                "0000000011111111222222223333333344444444555555556666666677777777",
            )?),
            ..Default::default()
        };
        let store = DbKeyStore::new(config)?;
        eprintln!("with_enc: {store:?}");

        let config = DbKeyStoreConfig {
            vfs: Some("memory".to_string()),
            ..Default::default()
        };
        let store = DbKeyStore::new(config)?;
        eprintln!("memory: {store:?}");
        Ok(())
    }

    #[test]
    fn uuid_v7_strings_are_lexicographically_increasing() {
        let mut uuids = Vec::new();
        for _ in 0..8 {
            uuids.push(new_uuid());
        }
        for pair in uuids.windows(2) {
            assert!(
                pair[0] < pair[1],
                "uuid v7 strings should be lexicographically increasing"
            );
        }
    }
}