rustango 0.60.0

//! A [`Cache`] backend that stores one row per key in the database.
//! The table is `cache_key`, `value` and `expires`,
//! with the same layout on PG, MySQL and SQLite. Expired rows are
//! removed on read; there is no background reaper.
//!
//! ## Quick start
//!
//! ```ignore
//! use std::sync::Arc;
//! use rustango::cache::{Cache, DatabaseCache};
//!
//! let cache = DatabaseCache::new(pool.clone(), "rustango_cache");
//! cache.ensure_table().await?;                  // one-time idempotent DDL
//! let cache: Arc<dyn Cache> = Arc::new(cache);
//! cache.set("greeting", "hello", None).await?;
//! ```
//!
//! ## Schema
//!
//! Same shape on every backend; only the column types vary so the
//! PRIMARY KEY survives across dialects:
//!
//! - Postgres: `cache_key TEXT PRIMARY KEY, value TEXT NOT NULL, expires BIGINT NOT NULL DEFAULT 0`
//! - MySQL:    `cache_key VARBINARY(255) PRIMARY KEY, value LONGTEXT NOT NULL, expires BIGINT NOT NULL DEFAULT 0`
//! - SQLite:   `cache_key TEXT PRIMARY KEY, value TEXT NOT NULL, expires INTEGER NOT NULL DEFAULT 0`
//!
//! `VARBINARY` makes MySQL compare keys byte for byte, as PG and SQLite do;
//! a default `VARCHAR` collation matched `user:1` to `User:1` (#1757).
//!
//! A key longer than 255 bytes is stored as its first 190 bytes plus
//! `#` and its SHA-256, so it fits MySQL's column and stays unique. A
//! shorter key that already ends that way is hashed too.
//!
//! `expires` holds Unix **milliseconds**; `0` means the entry never
//! expires. An entry is expired once now is *past* `expires`, not once
//! it reaches it — the same rule as [`super::InMemoryCache`] and
//! [`super::FileCache`]. A `get` or `exists` that lands on an expired
//! row deletes it and reports a miss.
//!
//! ## Why not a migration?
//!
//! The cache table does not follow the app's data. You can drop it,
//! recreate it, or keep it in another database.
//! [`DatabaseCache::ensure_table`] runs the right
//! `CREATE TABLE IF NOT EXISTS` at boot; `manage createcachetable`
//! does the same from the CLI.

use std::sync::atomic::{AtomicBool, Ordering};
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use async_trait::async_trait;

use super::{Cache, CacheError};
use crate::core::SqlValue;
use crate::sql::{raw_execute_pool, raw_query_pool, run_ddl_idempotent, Pool};

/// Longest key stored as-is: MySQL's `VARBINARY(255)` column.
const MAX_RAW_KEY_BYTES: usize = 255;

/// Bytes of a long key kept ahead of its hash, so prefix deletes still reach it.
const HASHED_HEAD_BYTES: usize = MAX_RAW_KEY_BYTES - 1 - 64;

/// Most rows one [`DatabaseCache::purge_expired`] statement deletes.
pub const PURGE_BATCH: u64 = 1000;

/// The `cache_key` column value: the key itself, or `{head}#{sha256 hex}`
/// when it is longer than [`MAX_RAW_KEY_BYTES`] or already ends like a hash,
/// so a raw key never equals a hashed one.
struct StoredKey(String);

impl StoredKey {
    fn new(key: &str) -> Self {
        use sha2::{Digest, Sha256};
        if key.len() <= MAX_RAW_KEY_BYTES && !ends_like_hash(key) {
            return Self(key.to_owned());
        }
        let digest = crate::hex::hex_encode(&Sha256::digest(key.as_bytes()));
        Self(format!("{}#{digest}", head(key)))
    }

    fn into_value(self) -> SqlValue {
        SqlValue::String(self.0)
    }
}

/// Whether `key` ends in `#` plus 64 lowercase hex, the hashed form's tail.
fn ends_like_hash(key: &str) -> bool {
    let b = key.as_bytes();
    b.len() >= 65
        && b[b.len() - 65] == b'#'
        && b[b.len() - 64..]
            .iter()
            .all(|c| c.is_ascii_digit() || (b'a'..=b'f').contains(c))
}

/// `s` cut to at most [`HASHED_HEAD_BYTES`], on a char boundary.
fn head(s: &str) -> &str {
    let mut end = HASHED_HEAD_BYTES.min(s.len());
    while !s.is_char_boundary(end) {
        end -= 1;
    }
    &s[..end]
}

/// Cache stored in a SQL table. Holds a [`Pool`] and a table name;
/// every operation runs one statement written for that dialect.
#[derive(Clone)]
pub struct DatabaseCache {
    pool: Pool,
    table: String,
}

impl DatabaseCache {
    /// Build a cache writing to `table` on `pool`. Call
    /// [`Self::ensure_table`] once at startup to create the table.
    #[must_use]
    pub fn new(pool: Pool, table: impl Into<String>) -> Self {
        Self {
            pool,
            table: table.into(),
        }
    }

    /// The configured table name.
    #[must_use]
    pub fn table(&self) -> &str {
        &self.table
    }

    /// Create the cache table if it is missing. Safe to call at every
    /// boot.
    ///
    /// # Errors
    /// [`CacheError::Connection`] when the table DDL fails, for example on a
    /// permission problem. A failed `expires` index only logs a warning.
    pub async fn ensure_table(&self) -> Result<(), CacheError> {
        let dialect = self.pool.dialect();
        let table = dialect.quote_ident(&self.table);
        let sql = match dialect.name() {
            "postgres" => format!(
                "CREATE TABLE IF NOT EXISTS {table} (\
                 cache_key TEXT PRIMARY KEY, \
                 value TEXT NOT NULL, \
                 expires BIGINT NOT NULL DEFAULT 0\
                 )"
            ),
            "mysql" => format!(
                "CREATE TABLE IF NOT EXISTS {table} (\
                 cache_key VARBINARY(255) PRIMARY KEY, \
                 value LONGTEXT NOT NULL, \
                 expires BIGINT NOT NULL DEFAULT 0\
                 )"
            ),
            // SQLite, and a reasonable default for anything else.
            _ => format!(
                "CREATE TABLE IF NOT EXISTS {table} (\
                 cache_key TEXT PRIMARY KEY, \
                 value TEXT NOT NULL, \
                 expires INTEGER NOT NULL DEFAULT 0\
                 )"
            ),
        };
        raw_execute_pool(&self.pool, &sql, vec![])
            .await
            .map_err(|e| CacheError::Connection(format!("ensure_table: {e}")))?;
        // `purge_expired` scans by `expires`; existing tables get it too.
        let index = dialect.quote_ident(&self.expires_index());
        let guard = if dialect.supports_create_index_if_not_exists() {
            "IF NOT EXISTS "
        } else {
            ""
        };
        // Best effort: a non-owner PG role or a MySQL role without INDEX still gets a cache.
        let ddl = format!("CREATE INDEX {guard}{index} ON {table} (expires)");
        if let Err(e) = run_ddl_idempotent(&self.pool, &ddl).await {
            static WARNED: AtomicBool = AtomicBool::new(false);
            if !WARNED.swap(true, Ordering::Relaxed) {
                tracing::warn!(
                    target: "rustango::cache",
                    table = %self.table,
                    error = %e,
                    "DatabaseCache: could not create the expires index; purge_expired will scan the table"
                );
            }
        }
        Ok(())
    }

    /// Name of the `expires` index. Postgres index names are per schema, so it carries the table.
    /// A long table name is cut and hashed to stay under PG's 63 and MySQL's 64 bytes.
    #[doc(hidden)]
    #[must_use]
    pub fn expires_index(&self) -> String {
        const MAX_IDENT: usize = 63;
        let plain = format!("{}_expires_idx", self.table);
        if plain.len() <= MAX_IDENT {
            return plain;
        }
        use sha2::{Digest, Sha256};
        let mut head = 40;
        while !self.table.is_char_boundary(head) {
            head -= 1;
        }
        let digest = crate::hex::hex_encode(&Sha256::digest(self.table.as_bytes()));
        format!("{}_{}_exp", &self.table[..head], &digest[..8])
    }

    /// Drop the cache table. Handy in tests. In production, run it
    /// from a manage verb rather than from app code.
    ///
    /// # Errors
    /// [`CacheError::Connection`] when the statement fails.
    pub async fn drop_table(&self) -> Result<(), CacheError> {
        let table = self.pool.dialect().quote_ident(&self.table);
        let sql = format!("DROP TABLE IF EXISTS {table}");
        raw_execute_pool(&self.pool, &sql, vec![])
            .await
            .map_err(|e| CacheError::Connection(format!("drop_table: {e}")))?;
        Ok(())
    }

    /// Delete every expired row now. `get` and `exists` already clean
    /// up rows they touch, so run this on a schedule to reclaim space
    /// from keys nobody reads.
    ///
    /// Returns the number of rows deleted. Rows with `expires = 0`
    /// have no TTL and are never touched. Deletes in batches of
    /// [`PURGE_BATCH`] rows, so no one statement holds locks on the whole table.
    ///
    /// # Errors
    /// [`CacheError::Connection`] when a DELETE fails.
    pub async fn purge_expired(&self) -> Result<u64, CacheError> {
        let dialect = self.pool.dialect();
        let table = dialect.quote_ident(&self.table);
        let p1 = dialect.placeholder(1);
        // Keep no-TTL rows. Use the same clock as `get`/`set` so both
        // agree on what "expired" means.
        let expired = format!("expires <> 0 AND expires < {p1}");
        // MySQL rejects LIMIT inside IN (…); PG has no DELETE … LIMIT.
        let sql = if dialect.name() == "mysql" {
            format!("DELETE FROM {table} WHERE {expired} LIMIT {PURGE_BATCH}")
        } else {
            format!(
                "DELETE FROM {table} WHERE cache_key IN \
                 (SELECT cache_key FROM {table} WHERE {expired} LIMIT {PURGE_BATCH})"
            )
        };
        let now = Self::now_unix_ms();
        let mut total = 0;
        loop {
            let n = raw_execute_pool(&self.pool, &sql, vec![SqlValue::I64(now)])
                .await
                .map_err(|e| CacheError::Connection(format!("purge_expired: {e}")))?;
            total += n;
            if n < PURGE_BATCH {
                return Ok(total);
            }
        }
    }

    /// Drop `key` only if it is still expired, so a `set` racing the
    /// read is not lost (#1906).
    async fn delete_if_expired(&self, key: &str) -> Result<u64, CacheError> {
        let dialect = self.pool.dialect();
        let table = dialect.quote_ident(&self.table);
        let (p1, p2) = (dialect.placeholder(1), dialect.placeholder(2));
        let sql = format!(
            "DELETE FROM {table} WHERE cache_key = {p1} AND expires <> 0 AND expires < {p2}"
        );
        let binds = vec![
            StoredKey::new(key).into_value(),
            SqlValue::I64(Self::now_unix_ms()),
        ];
        raw_execute_pool(&self.pool, &sql, binds)
            .await
            .map_err(|e| CacheError::Connection(format!("get: {e}")))
    }

    /// Unix epoch in **milliseconds**. Seconds are too coarse: a
    /// `set` at `HH:MM:SS.999` read back 2 ms later would truncate to
    /// two different seconds, so a 1-second TTL looked expired at
    /// once.
    fn now_unix_ms() -> i64 {
        SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map(|d| i64::try_from(d.as_millis()).unwrap_or(i64::MAX))
            .unwrap_or(0)
    }

    fn expires_for(ttl: Option<Duration>) -> i64 {
        ttl.map(|d| {
            let ms = i64::try_from(d.as_millis()).unwrap_or(i64::MAX);
            Self::now_unix_ms().saturating_add(ms)
        })
        .unwrap_or(0)
    }
}

#[async_trait]
impl Cache for DatabaseCache {
    async fn get(&self, key: &str) -> Result<Option<String>, CacheError> {
        let dialect = self.pool.dialect();
        let table = dialect.quote_ident(&self.table);
        let p1 = dialect.placeholder(1);
        let sql = format!("SELECT value, expires FROM {table} WHERE cache_key = {p1} LIMIT 1");
        let rows: Vec<(String, i64)> =
            raw_query_pool(&sql, vec![StoredKey::new(key).into_value()], &self.pool)
                .await
                .map_err(|e| CacheError::Connection(format!("get: {e}")))?;
        let Some((value, expires)) = rows.into_iter().next() else {
            return Ok(None);
        };
        // `>`, not `>=`, so an entry lives its full stated duration.
        // The other backends and `purge_expired` use `>` as well.
        if expires != 0 && Self::now_unix_ms() > expires {
            let _ = self.delete_if_expired(key).await;
            return Ok(None);
        }
        Ok(Some(value))
    }

    async fn set(&self, key: &str, value: &str, ttl: Option<Duration>) -> Result<(), CacheError> {
        let dialect = self.pool.dialect();
        let table = dialect.quote_ident(&self.table);
        let p1 = dialect.placeholder(1);
        let p2 = dialect.placeholder(2);
        let p3 = dialect.placeholder(3);
        let expires = Self::expires_for(ttl);
        // Per-dialect upsert. MySQL has no ON CONFLICT clause.
        let sql = match dialect.name() {
            "mysql" => format!(
                "INSERT INTO {table} (cache_key, value, expires) \
                 VALUES ({p1}, {p2}, {p3}) \
                 ON DUPLICATE KEY UPDATE value = VALUES(value), expires = VALUES(expires)"
            ),
            _ => format!(
                "INSERT INTO {table} (cache_key, value, expires) \
                 VALUES ({p1}, {p2}, {p3}) \
                 ON CONFLICT (cache_key) DO UPDATE SET value = EXCLUDED.value, expires = EXCLUDED.expires"
            ),
        };
        raw_execute_pool(
            &self.pool,
            &sql,
            vec![
                StoredKey::new(key).into_value(),
                SqlValue::String(value.to_owned()),
                SqlValue::I64(expires),
            ],
        )
        .await
        .map_err(|e| CacheError::Connection(format!("set: {e}")))?;
        Ok(())
    }

    /// Atomic set-if-absent. `DistributedLock` needs this: the trait
    /// default checks and then writes, so two racers could both get
    /// `Ok(true)` and both run the guarded body.
    ///
    /// An expired row must still be takeable, so this is not a plain
    /// `INSERT`. It takes the row when it is absent *or* expired, and
    /// says whether it did. A row with `expires = 0` never expires, so
    /// a live permanent entry is never stolen.
    ///
    /// The database does the test-and-set under row locks, so exactly
    /// one racer wins: once the winner moves `expires` forward, the
    /// loser's `WHERE` no longer matches.
    async fn add(&self, key: &str, value: &str, ttl: Option<Duration>) -> Result<bool, CacheError> {
        let dialect = self.pool.dialect();
        let table = dialect.quote_ident(&self.table);
        let p1 = dialect.placeholder(1);
        let p2 = dialect.placeholder(2);
        let p3 = dialect.placeholder(3);
        let p4 = dialect.placeholder(4);
        let expires = Self::expires_for(ttl);
        let now = Self::now_unix_ms();

        // Bind args in the order the placeholders appear in the text.
        // MySQL and SQLite use positional `?`. Postgres' `$n` would
        // accept any order, which is how a MySQL-only mismatch hides
        // until a MySQL test runs.
        if dialect.name() == "mysql" {
            // MySQL's `ON DUPLICATE KEY UPDATE` takes no WHERE, so
            // this needs two statements. Still race-free: the INSERT
            // covers the absent case, and the UPDATE's own predicate
            // is the compare-and-swap for the expired one.
            let inserted = raw_execute_pool(
                &self.pool,
                &format!(
                    "INSERT IGNORE INTO {table} (cache_key, value, expires) \
                     VALUES ({p1}, {p2}, {p3})"
                ),
                vec![
                    StoredKey::new(key).into_value(),
                    SqlValue::String(value.to_owned()),
                    SqlValue::I64(expires),
                ],
            )
            .await
            .map_err(|e| CacheError::Connection(format!("add: {e}")))?;
            if inserted == 1 {
                return Ok(true);
            }
            let took = raw_execute_pool(
                &self.pool,
                &format!(
                    "UPDATE {table} SET value = {p1}, expires = {p2} \
                     WHERE cache_key = {p3} AND expires <> 0 AND expires <= {p4}"
                ),
                vec![
                    SqlValue::String(value.to_owned()),
                    SqlValue::I64(expires),
                    StoredKey::new(key).into_value(),
                    SqlValue::I64(now),
                ],
            )
            .await
            .map_err(|e| CacheError::Connection(format!("add: {e}")))?;
            return Ok(took == 1);
        }

        // Postgres and SQLite support `ON CONFLICT … DO UPDATE …
        // WHERE`, so one statement does it: insert, or overwrite only
        // an expired row. A failed predicate writes nothing and
        // reports 0 rows.
        let took = raw_execute_pool(
            &self.pool,
            &format!(
                "INSERT INTO {table} (cache_key, value, expires) \
                 VALUES ({p1}, {p2}, {p3}) \
                 ON CONFLICT (cache_key) DO UPDATE \
                 SET value = EXCLUDED.value, expires = EXCLUDED.expires \
                 WHERE {table}.expires <> 0 AND {table}.expires <= {p4}"
            ),
            // Textual placeholder order; see the note above.
            vec![
                StoredKey::new(key).into_value(),
                SqlValue::String(value.to_owned()),
                SqlValue::I64(expires),
                SqlValue::I64(now),
            ],
        )
        .await
        .map_err(|e| CacheError::Connection(format!("add: {e}")))?;
        Ok(took == 1)
    }

    /// One upsert, so parallel callers never lose a count (#1871). The TTL
    /// is set on insert only; an expired or non-integer value restarts at `by`.
    /// Overflow is an error, not a saturation.
    async fn incr(&self, key: &str, by: i64, ttl: Option<Duration>) -> Result<i64, CacheError> {
        let dialect = self.pool.dialect();
        let table = dialect.quote_ident(&self.table);
        let (p1, p2, p3, p4, p5, p6) = (
            dialect.placeholder(1),
            dialect.placeholder(2),
            dialect.placeholder(3),
            dialect.placeholder(4),
            dialect.placeholder(5),
            dialect.placeholder(6),
        );
        let now = Self::now_unix_ms();
        // Binds in text order: MySQL and SQLite are positional.
        let mut binds = vec![
            StoredKey::new(key).into_value(),
            SqlValue::String(by.to_string()),
            SqlValue::I64(Self::expires_for(ttl)),
            SqlValue::I64(now),
            SqlValue::I64(now),
        ];
        let err = |e: crate::sql::ExecError| CacheError::Connection(format!("incr: {e}"));
        let mysql = dialect.name() == "mysql";
        let excluded = |col: &str| {
            if mysql {
                format!("VALUES({col})")
            } else {
                format!("EXCLUDED.{col}")
            }
        };
        let (new_value, new_expires) = (excluded("value"), excluded("expires"));
        let expired = |p: &str| format!("{table}.expires <> 0 AND {table}.expires < {p}");
        // A strict-mode MySQL CAST of a non-integer is an error, not 0.
        let (not_int, int) = match dialect.name() {
            "mysql" => (
                format!("{table}.value NOT REGEXP '^-?[0-9]{{1,18}}$'"),
                "SIGNED",
            ),
            "postgres" => (format!("{table}.value !~ '^-?[0-9]{{1,18}}$'"), "BIGINT"),
            _ => (
                format!("CAST(CAST({table}.value AS INTEGER) AS TEXT) <> {table}.value"),
                "INTEGER",
            ),
        };
        let text = if mysql { "CHAR" } else { "TEXT" };
        // MySQL applies the assignments left to right, so `value` goes
        // first while `expires` still holds the old row's value.
        let set = format!(
            "value = CASE WHEN ({}) OR {not_int} THEN {new_value} \
             ELSE CAST(CAST({table}.value AS {int}) + CAST({new_value} AS {int}) AS {text}) END, \
             expires = CASE WHEN {} THEN {new_expires} ELSE {table}.expires END",
            expired(&p4),
            expired(&p5),
        );
        let insert =
            format!("INSERT INTO {table} (cache_key, value, expires) VALUES ({p1}, {p2}, {p3})");
        let value: String = if dialect.name() == "mysql" {
            // No RETURNING: read back in the same transaction, which still
            // holds the row lock the upsert took.
            let mut tx = crate::sql::transaction_pool(&self.pool)
                .await
                .map_err(err)?;
            crate::sql::raw_execute_tx(
                &mut tx,
                &format!("{insert} ON DUPLICATE KEY UPDATE {set}"),
                binds,
            )
            .await
            .map_err(err)?;
            let rows: Vec<(String,)> = crate::sql::raw_query_tx(
                &mut tx,
                &format!("SELECT value FROM {table} WHERE cache_key = {p1}"),
                vec![StoredKey::new(key).into_value()],
            )
            .await
            .map_err(err)?;
            tx.commit().await.map_err(|e| err(e.into()))?;
            rows.into_iter().next().map(|(v,)| v).unwrap_or_default()
        } else {
            // SQLite turns an overflowing sum into a REAL; skip that update
            // instead, and report it as PG and MySQL do.
            let guard = if dialect.name() == "sqlite" {
                binds.push(SqlValue::I64(now));
                format!(
                    " WHERE ({}) OR {not_int} OR typeof(CAST({table}.value AS INTEGER) + \
                     CAST({new_value} AS INTEGER)) = 'integer'",
                    expired(&p6)
                )
            } else {
                String::new()
            };
            let rows: Vec<(String,)> = raw_query_pool(
                &format!(
                    "{insert} ON CONFLICT (cache_key) DO UPDATE SET {set}{guard} RETURNING value"
                ),
                binds,
                &self.pool,
            )
            .await
            .map_err(err)?;
            let Some((v,)) = rows.into_iter().next() else {
                return Err(CacheError::Connection(
                    "incr: integer out of range (i64 overflow)".to_owned(),
                ));
            };
            v
        };
        value
            .parse()
            .map_err(|_| CacheError::Connection(format!("incr: non-integer result {value:?}")))
    }

    async fn delete(&self, key: &str) -> Result<(), CacheError> {
        let dialect = self.pool.dialect();
        let table = dialect.quote_ident(&self.table);
        let p1 = dialect.placeholder(1);
        let sql = format!("DELETE FROM {table} WHERE cache_key = {p1}");
        raw_execute_pool(&self.pool, &sql, vec![StoredKey::new(key).into_value()])
            .await
            .map_err(|e| CacheError::Connection(format!("delete: {e}")))?;
        Ok(())
    }

    async fn exists(&self, key: &str) -> Result<bool, CacheError> {
        Ok(self.get(key).await?.is_some())
    }

    async fn clear(&self) -> Result<(), CacheError> {
        let dialect = self.pool.dialect();
        let table = dialect.quote_ident(&self.table);
        let sql = format!("DELETE FROM {table}");
        raw_execute_pool(&self.pool, &sql, vec![])
            .await
            .map_err(|e| CacheError::Connection(format!("clear: {e}")))?;
        Ok(())
    }

    /// Prefix delete with `LIKE 'prefix%'`. The keys are a real
    /// column, so no whole-table clear is needed.
    ///
    /// `%` and `_` are LIKE wildcards, so a prefix containing either
    /// has to be escaped or it would sweep a neighbour's rows.
    ///
    /// The escape character is **`!`, not `\`**. A backslash is not
    /// portable: `ESCAPE '\'` is a syntax error on MySQL but valid on
    /// Postgres. `!` needs no escaping on any of the three, so one
    /// statement works everywhere.
    ///
    /// SQLite's `LIKE` ignores ASCII case, so it matches with
    /// `instr(..) = 1` instead; the prefix stays exact on every backend.
    async fn delete_prefix(&self, prefix: &str) -> Result<(), CacheError> {
        let dialect = self.pool.dialect();
        let table = dialect.quote_ident(&self.table);
        let p = dialect.placeholder(1);
        // Use the crate-wide escaper so the escape character cannot
        // drift between the cache and the ORM's LIKE lookups.
        // A hashed key keeps only its head, so a longer prefix matches on that
        // head: this may drop extra entries, never fewer.
        let (sql, arg) = if dialect.name() == "sqlite" {
            (
                format!("DELETE FROM {table} WHERE instr(cache_key, {p}) = 1"),
                head(prefix).to_owned(),
            )
        } else {
            (
                format!(
                    "DELETE FROM {table} WHERE cache_key LIKE {p}{}",
                    crate::core::LIKE_ESCAPE_CLAUSE
                ),
                format!("{}%", crate::core::escape_like(head(prefix))),
            )
        };
        raw_execute_pool(&self.pool, &sql, vec![SqlValue::String(arg)])
            .await
            .map_err(|e| CacheError::Connection(format!("delete_prefix: {e}")))?;
        Ok(())
    }
}

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

    /// No TTL means `expires = 0`, which never expires.
    #[test]
    fn expires_for_zero_when_no_ttl() {
        assert_eq!(DatabaseCache::expires_for(None), 0);
    }

    /// The expired-read cleanup must spare a value a racing `set` just wrote (#1906).
    #[cfg(feature = "sqlite")]
    #[tokio::test]
    async fn expired_cleanup_spares_a_fresh_value() {
        let pool = Pool::connect("sqlite::memory:").await.unwrap();
        let cache = DatabaseCache::new(pool, "c");
        cache.ensure_table().await.unwrap();
        cache.set("k", "fresh", None).await.unwrap();
        assert_eq!(cache.delete_if_expired("k").await.unwrap(), 0);
        cache
            .set("k", "fresh", Some(Duration::from_secs(60)))
            .await
            .unwrap();
        assert_eq!(cache.delete_if_expired("k").await.unwrap(), 0);
        assert_eq!(cache.get("k").await.unwrap().as_deref(), Some("fresh"));
        cache
            .set("k", "old", Some(Duration::from_millis(1)))
            .await
            .unwrap();
        tokio::time::sleep(Duration::from_millis(5)).await;
        assert_eq!(cache.delete_if_expired("k").await.unwrap(), 1);
    }

    #[test]
    fn short_keys_are_stored_unchanged() {
        let k = "x".repeat(MAX_RAW_KEY_BYTES);
        assert_eq!(StoredKey::new(&k).0, k);
    }

    #[test]
    fn long_keys_fit_the_column_and_stay_distinct() {
        // Multi-byte chars around the cut must not split a char.
        let base = "é".repeat(200);
        let a = StoredKey::new(&format!("{base}a")).0;
        let b = StoredKey::new(&format!("{base}b")).0;
        assert!(a.len() <= MAX_RAW_KEY_BYTES, "{}", a.len());
        assert_ne!(a, b);
        assert!(a.starts_with(head(&base)));
    }

    #[test]
    fn a_raw_key_never_equals_a_hashed_one() {
        let long = "x".repeat(300);
        let hashed = StoredKey::new(&long).0;
        assert!(hashed.len() <= MAX_RAW_KEY_BYTES);
        // Writing the hashed form as a raw key must not reach the long key's row.
        assert_ne!(StoredKey::new(&hashed).0, hashed);
    }

    #[test]
    fn expires_for_offsets_from_now() {
        // A 60-second TTL adds 60_000 ms to the clock reading.
        let before = DatabaseCache::now_unix_ms();
        let ts = DatabaseCache::expires_for(Some(Duration::from_secs(60)));
        let after = DatabaseCache::now_unix_ms();
        assert!(
            ts >= before + 60_000 && ts <= after + 60_000,
            "expected expires in [{before}+60_000, {after}+60_000], got {ts}"
        );
    }
}