rullst-orm 13.0.0-alpha.1

use sha2::{Digest, Sha256};

use crate::{Error, Orm, RullstValue};

mod model_hash;
#[doc(hidden)]
pub use model_hash::{increment_model_hash, model_hash_key, read_model_hash, write_model_hash};

/// Version 4 indexes entries in a sorted set; version 3 entries, indexed in a
/// plain set, are never read again and expire through their TTL.
const KEY_PREFIX: &str = "rullst:orm:cache:v4:";
const MAX_NAMESPACE_LEN: usize = 64;
const MAX_INVALIDATION_KEYS: usize = 10_000;
/// Suffix of the per-table sorted set that indexes the table's cached
/// entries. It is not hexadecimal, so it can never equal an entry's digest
/// segment.
const INDEX_SUFFIX: &str = "index";

/// Lua prelude shared by the index scripts: `now_ms` is the Redis server
/// clock in milliseconds, the unit of key expiry, and index members scored
/// below it belong to entries that have already expired. Redis 5+ replicates
/// script effects, so a write after `TIME` is allowed; `replicate_commands`
/// keeps that true on older servers and is a no-op where it is deprecated.
const NOW_PRELUDE: &str = r"
if redis.replicate_commands then redis.replicate_commands() end
local clock = redis.call('TIME')
local now_ms = tonumber(clock[1]) * 1000 + math.floor(tonumber(clock[2]) / 1000)
redis.call('ZREMRANGEBYSCORE', KEYS[#KEYS], '-inf', string.format('(%.0f', now_ms))
";

/// Stores an entry, scores its key in the table index by the entry's expiry
/// time, prunes members of already-expired entries and extends the index's
/// lifetime to the longest entry TTL, as one atomic step. `KEYS[2]` is the
/// index (the prelude prunes `KEYS[#KEYS]`).
const STORE_BODY: &str = r"
local ttl = tonumber(ARGV[2])
redis.call('SET', KEYS[1], ARGV[1], 'EX', ARGV[2])
redis.call('ZADD', KEYS[2], string.format('%.0f', now_ms + ttl * 1000), KEYS[1])
if redis.call('TTL', KEYS[2]) < ttl then
  redis.call('EXPIRE', KEYS[2], ARGV[2])
end
return 1
";

/// Removes at most `ARGV[1]` live indexed entries together with their index
/// members, after pruning the members of expired entries, and returns the
/// number removed and the number of live members left. Keys added
/// concurrently either leave in this batch or stay indexed.
const INVALIDATE_BODY: &str = r"
local limit = tonumber(ARGV[1])
local removed = 0
if limit > 0 then
  local keys = redis.call('ZRANGE', KEYS[1], 0, limit - 1)
  removed = #keys
  if removed > 0 then
    redis.call('ZREM', KEYS[1], unpack(keys))
    redis.call('UNLINK', unpack(keys))
  end
end
return {removed, redis.call('ZCARD', KEYS[1])}
";
const INVALIDATION_BATCH: usize = 500;

pub(crate) fn validate_namespace(namespace: &str) -> Result<(), Error> {
    if namespace.is_empty()
        || namespace.len() > MAX_NAMESPACE_LEN
        || !namespace
            .bytes()
            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.'))
    {
        return Err(Error::Validation(
            "Redis cache namespace must contain 1-64 ASCII letters, digits, '.', '-' or '_'"
                .to_string(),
        ));
    }
    Ok(())
}

/// Builds the versioned key of a `.remember(...)` query in the active tenant
/// scope's partition (`global` outside `with_tenant`).
pub fn query_key(table: &str, query: &str, bindings: &[RullstValue]) -> Result<String, Error> {
    let namespace = Orm::redis_cache_namespace()?;
    let tenant = crate::tenant::get_tenant_id();
    build_key(namespace, tenant.as_ref(), table, query, bindings)
}

/// Builds the key of a generated `.remember(...)` read. Only a model with a
/// tenant column partitions its entries by the active tenant; any other model
/// keeps one `global` copy whatever `with_tenant` scope is active, so a write
/// from any scope invalidates it.
#[doc(hidden)]
pub fn query_key_for_model(
    table: &str,
    tenant_scoped: bool,
    query: &str,
    bindings: &[RullstValue],
) -> Result<String, Error> {
    let namespace = Orm::redis_cache_namespace()?;
    let tenant = tenant_scoped.then(crate::tenant::get_tenant_id).flatten();
    build_key(namespace, tenant.as_ref(), table, query, bindings)
}

fn build_key(
    namespace: &str,
    tenant: Option<&RullstValue>,
    table: &str,
    query: &str,
    bindings: &[RullstValue],
) -> Result<String, Error> {
    validate_namespace(namespace)?;

    let mut digest = Sha256::new();
    update_field(&mut digest, 1, namespace.as_bytes());
    if let Some(value) = tenant {
        update_value(&mut digest, value);
    } else {
        update_field(&mut digest, 2, b"global");
    }
    update_field(&mut digest, 3, table.as_bytes());
    update_field(&mut digest, 4, query.as_bytes());
    for binding in bindings {
        update_value(&mut digest, binding);
    }

    Ok(format!(
        "{}:{}",
        table_prefix(namespace, tenant, table)?,
        hex_digest(&digest.finalize())
    ))
}

/// Common prefix of one namespace, tenant scope and table's cache keys.
fn table_prefix(
    namespace: &str,
    tenant: Option<&RullstValue>,
    table: &str,
) -> Result<String, Error> {
    validate_namespace(namespace)?;
    Ok(format!(
        "{KEY_PREFIX}{namespace}:{}:table-{}",
        scope_segment(tenant),
        digest_bytes(table.as_bytes())
    ))
}

/// The index of the table that a generated entry key belongs to.
fn entry_index_key(cache_key: &str) -> Result<String, Error> {
    match cache_key.rsplit_once(':') {
        Some((prefix, digest))
            if prefix.starts_with(KEY_PREFIX)
                && prefix.contains(":table-")
                && digest.len() == 64
                && digest.bytes().all(|byte| byte.is_ascii_hexdigit()) =>
        {
            Ok(format!("{prefix}:{INDEX_SUFFIX}"))
        }
        _ => Err(Error::Validation(
            "query cache key is not a generated entry key".to_string(),
        )),
    }
}

/// Stores one generated query result and indexes its key under its table,
/// so a committed write can invalidate the table without scanning Redis.
#[doc(hidden)]
pub async fn store_entry(cache_key: &str, payload: &str, ttl_seconds: u64) -> Result<(), Error> {
    if ttl_seconds == 0 {
        return Err(Error::Validation(
            "query cache TTL must be greater than zero".to_string(),
        ));
    }
    let index = entry_index_key(cache_key)?;
    let mut connection = Orm::redis_manager()?;
    let _: i64 = crate::_redis::cmd("EVAL")
        .arg(format!("{NOW_PRELUDE}{STORE_BODY}"))
        .arg(2)
        .arg(cache_key)
        .arg(&index)
        .arg(payload)
        .arg(ttl_seconds)
        .query_async(&mut connection)
        .await?;
    Ok(())
}

/// Deletes every indexed query-cache entry for a table.
///
/// An unconfigured Redis adapter is a no-op because model writes do not require
/// caching. Once Redis is configured, transport errors fail visibly. The work
/// follows the table's own index, never the rest of the Redis keyspace, and is
/// capped at `MAX_INVALIDATION_KEYS` entries per call; entries beyond the cap
/// stay indexed for the next write and otherwise expire through their TTL, as
/// do entries written by earlier versions that were never indexed.
pub async fn invalidate_table(table: &str) -> Result<usize, Error> {
    let Ok(namespace) = Orm::redis_cache_namespace() else {
        return Ok(0);
    };
    let tenant = crate::tenant::get_tenant_id();
    drain_indexes(&[index_key(namespace, tenant.as_ref(), table)?]).await
}

/// Invalidates a written model's cached reads after commit: the `global`
/// partition (every read of a model without a tenant column, and `unscoped()`
/// reads outside `with_tenant`) and, when `tenant` is the writing tenant of a
/// tenant-scoped model, that tenant's partition. The per-write key cap of
/// [`invalidate_table`] covers both partitions.
#[doc(hidden)]
pub async fn invalidate_model_table(
    table: &str,
    tenant: Option<RullstValue>,
) -> Result<usize, Error> {
    let Ok(namespace) = Orm::redis_cache_namespace() else {
        return Ok(0);
    };
    let mut indexes = vec![index_key(namespace, None, table)?];
    if let Some(tenant) = tenant.as_ref() {
        indexes.push(index_key(namespace, Some(tenant), table)?);
    }
    drain_indexes(&indexes).await
}

/// Removes and unlinks the live entries of each index in batches, at most
/// `MAX_INVALIDATION_KEYS` in total; entries beyond the cap stay indexed.
/// Members whose entries already expired are pruned without counting toward
/// the cap.
async fn drain_indexes(indexes: &[String]) -> Result<usize, Error> {
    let script = format!("{NOW_PRELUDE}{INVALIDATE_BODY}");
    let mut connection = Orm::redis_manager()?;
    let mut deleted = 0_usize;
    for index in indexes {
        loop {
            let batch = INVALIDATION_BATCH.min(MAX_INVALIDATION_KEYS.saturating_sub(deleted));
            let (removed, remaining): (usize, usize) = crate::_redis::cmd("EVAL")
                .arg(&script)
                .arg(1)
                .arg(index)
                .arg(batch)
                .query_async(&mut connection)
                .await?;
            deleted = deleted.saturating_add(removed);
            if remaining == 0 {
                break;
            }
            if deleted >= MAX_INVALIDATION_KEYS {
                return Err(Error::CacheError(format!(
                    "table cache invalidation exceeded {MAX_INVALIDATION_KEYS} keys; the rest expire by TTL"
                )));
            }
        }
    }
    Ok(deleted)
}

/// The sorted set indexing one namespace, tenant scope and table's cache
/// entries.
fn index_key(namespace: &str, tenant: Option<&RullstValue>, table: &str) -> Result<String, Error> {
    Ok(format!(
        "{}:{INDEX_SUFFIX}",
        table_prefix(namespace, tenant, table)?
    ))
}

fn scope_segment(tenant: Option<&RullstValue>) -> String {
    match tenant {
        Some(value) => format!("tenant-{}", short_digest(value)),
        None => "global".to_string(),
    }
}

fn short_digest(value: &RullstValue) -> String {
    let mut digest = Sha256::new();
    digest.update(b"rullst:orm:tenant:v1");
    update_value(&mut digest, value);
    let output = digest.finalize();
    hex_digest(&output[..16])
}

fn digest_bytes(bytes: &[u8]) -> String {
    let mut digest = Sha256::new();
    digest.update(b"rullst:orm:cache-table:v1");
    digest.update((bytes.len() as u64).to_be_bytes());
    digest.update(bytes);
    hex_digest(&digest.finalize()[..16])
}

fn update_value(digest: &mut Sha256, value: &RullstValue) {
    match value {
        RullstValue::String(value) => update_field(digest, 10, value.as_bytes()),
        RullstValue::Int(value) => update_field(digest, 11, &value.to_be_bytes()),
        RullstValue::Float(value) => update_field(digest, 12, &value.to_bits().to_be_bytes()),
        RullstValue::Bool(value) => update_field(digest, 13, &[*value as u8]),
    }
}

fn update_field(digest: &mut Sha256, tag: u8, bytes: &[u8]) {
    digest.update([tag]);
    digest.update((bytes.len() as u64).to_be_bytes());
    digest.update(bytes);
}

fn hex_digest(bytes: &[u8]) -> String {
    const HEX: &[u8; 16] = b"0123456789abcdef";
    let mut encoded = String::with_capacity(bytes.len() * 2);
    for byte in bytes {
        encoded.push(char::from(HEX[usize::from(byte >> 4)]));
        encoded.push(char::from(HEX[usize::from(byte & 0x0f)]));
    }
    encoded
}

#[cfg(test)]
mod tests {
    use super::{build_key, entry_index_key, index_key};
    use crate::RullstValue;

    #[test]
    fn entry_keys_resolve_to_their_table_index() {
        let tenant = RullstValue::Int(7);
        let entry =
            build_key("academy", Some(&tenant), "users", "SELECT *", &[]).expect("entry key");
        let index = index_key("academy", Some(&tenant), "users").expect("index key");
        assert_eq!(entry_index_key(&entry).expect("entry index"), index);
        assert_ne!(index, entry);
        for foreign in [
            "",
            "session:abc",
            "rullst:orm:cache:v4:academy:global:table-x:not-a-digest",
            index.as_str(),
        ] {
            assert!(entry_index_key(foreign).is_err(), "{foreign}");
        }
    }

    #[test]
    fn keys_are_versioned_deterministic_and_domain_separated() {
        let bindings = vec![RullstValue::String("12".to_string())];
        let first =
            build_key("academy", None, "users", "SELECT *", &bindings).expect("build cache key");
        let repeated =
            build_key("academy", None, "users", "SELECT *", &bindings).expect("repeat cache key");
        let typed = build_key(
            "academy",
            None,
            "users",
            "SELECT *",
            &[RullstValue::Int(12)],
        )
        .expect("typed cache key");

        assert_eq!(first, repeated);
        assert!(first.starts_with("rullst:orm:cache:v4:academy:global:table-"));
        assert_ne!(first, typed);
    }

    #[test]
    fn table_segment_is_explicit_but_does_not_expose_the_table_name() {
        let users =
            build_key("academy", None, "private_users", "SELECT *", &[]).expect("users cache key");
        let posts =
            build_key("academy", None, "private_posts", "SELECT *", &[]).expect("posts cache key");

        assert_ne!(users, posts);
        assert!(!users.contains("private_users"));
        assert!(!posts.contains("private_posts"));
    }

    #[test]
    fn tenant_and_application_namespaces_cannot_share_keys_or_leak_ids() {
        let tenant = RullstValue::String("private-school-name".to_string());
        let first = build_key("academy", Some(&tenant), "users", "SELECT *", &[])
            .expect("tenant cache key");
        let other_tenant = build_key(
            "academy",
            Some(&RullstValue::String("other".to_string())),
            "users",
            "SELECT *",
            &[],
        )
        .expect("other tenant cache key");
        let other_application = build_key("billing", Some(&tenant), "users", "SELECT *", &[])
            .expect("other application cache key");

        assert_ne!(first, other_tenant);
        assert_ne!(first, other_application);
        assert!(!first.contains("private-school-name"));
    }

    #[test]
    fn invalid_namespaces_fail_closed() {
        for invalid in ["", "has spaces", "tenant:escape", "../escape"] {
            assert!(build_key(invalid, None, "users", "SELECT *", &[]).is_err());
        }
        assert!(build_key(&"a".repeat(65), None, "users", "SELECT *", &[]).is_err());
    }

    #[test]
    fn invalidation_is_limited_to_the_active_opaque_tenant_scope() {
        let tenant = RullstValue::String("private-school-name".to_string());
        let tenant_index =
            index_key("academy", Some(&tenant), "users").expect("tenant invalidation index");
        let global_index = index_key("academy", None, "users").expect("global invalidation index");

        assert_ne!(tenant_index, global_index);
        assert!(tenant_index.contains(":tenant-"));
        assert!(global_index.contains(":global:"));
        assert!(!tenant_index.contains("private-school-name"));
        assert!(
            !tenant_index.contains('*'),
            "invalidation never uses a key pattern"
        );
    }
}