nodedb 0.3.0-beta.1

// SPDX-License-Identifier: BUSL-1.1

//! Hierarchical rate limiter: per-user → per-org → per-tenant → per-database.
//!
//! Each identity gets a token bucket. Requests consume tokens based on
//! endpoint cost multipliers. When empty, requests are rejected with 429.
//!
//! Hierarchy: per-key → per-user → per-org → per-tenant → per-database.
//! A request is allowed only if ALL applicable buckets have tokens.
//! Most-specific bucket that denies determines the error kind.
//!
//! ## Lock-poisoning policy
//!
//! The `RwLock`-guarded maps in this module hold owned `TokenBucket` values
//! whose internal state is a small struct of atomics + a refill timestamp.
//! Bucket mutations are individually consistent and do not span multiple
//! map operations. A panic in an unrelated request handler therefore cannot
//! corrupt the contents of these maps; only the `RwLock`'s poison flag is
//! set. Recovering via `unwrap_or_else(|p| p.into_inner())` keeps the rate
//! limiter live in the face of a one-off panic — the alternative (every
//! subsequent request returning a poisoning error) would itself be a
//! denial-of-service. Revisit if bucket mutation ever becomes a multi-step
//! protocol that can leave a bucket half-updated.

use std::collections::HashMap;
use std::sync::RwLock;

use tracing::debug;

use nodedb_types::{DatabaseId, TenantId};

use super::bucket::TokenBucket;
use super::config::RateLimitConfig;

/// Rate limit check result.
pub struct RateLimitResult {
    /// Whether the request is allowed.
    pub allowed: bool,
    /// Remaining tokens in the most constrained bucket.
    pub remaining: u64,
    /// Total limit of the most constrained bucket.
    pub limit: u64,
    /// Seconds until reset (0 if allowed).
    pub retry_after_secs: u64,
}

/// Parameters for a scoped rate-limit check that includes tenant and
/// database buckets in addition to the user/org hierarchy.
///
/// All fields are optional: set `None` to skip the corresponding bucket.
pub struct QuotaCheckParams {
    /// Tenant-scoped QPS cap (`tenant.quota.max_qps`). `0` or `None` = no cap.
    pub tenant_max_qps: Option<u64>,
    /// Database-scoped QPS cap (`database.quota.max_qps`). `0` or `None` = no cap.
    pub database_max_qps: Option<u64>,
    /// Tenant identifier (used as the bucket key when `tenant_max_qps` is set).
    pub tenant_id: TenantId,
    /// Database identifier (used as the bucket key when `database_max_qps` is set).
    pub database_id: DatabaseId,
}

/// Result of a pre-authentication login rate-limit check.
pub enum LoginRateLimitOutcome {
    /// Both the IP and user buckets have tokens remaining — proceed with auth.
    Allowed,
    /// The per-IP bucket was exhausted.
    IpExceeded,
    /// The per-username bucket was exhausted.
    UserExceeded,
}

/// Hierarchical rate limiter.
pub struct RateLimiter {
    config: RateLimitConfig,
    /// Per-identity buckets. Key = identity key (user_id, api_key_id, org_id).
    buckets: RwLock<HashMap<String, TokenBucket>>,
    /// Total rejection counter for Prometheus metrics.
    rejections_total: std::sync::atomic::AtomicU64,
    /// Maximum login attempts per IP per minute (0 = disabled).
    login_ip_cap: std::sync::atomic::AtomicU64,
    /// Maximum login attempts per username per minute (0 = disabled).
    login_user_cap: std::sync::atomic::AtomicU64,
}

impl RateLimiter {
    pub fn new(config: RateLimitConfig) -> Self {
        Self {
            config,
            buckets: RwLock::new(HashMap::new()),
            rejections_total: std::sync::atomic::AtomicU64::new(0),
            login_ip_cap: std::sync::atomic::AtomicU64::new(30),
            login_user_cap: std::sync::atomic::AtomicU64::new(10),
        }
    }

    /// Update the per-IP and per-username login attempt capacities.
    ///
    /// Takes effect for new token buckets created after this call.
    /// Existing in-flight buckets retain their original capacity.
    /// Called once at startup from server configuration.
    pub fn set_login_capacities(&self, ip_cap: u64, user_cap: u64) {
        self.login_ip_cap
            .store(ip_cap, std::sync::atomic::Ordering::Relaxed);
        self.login_user_cap
            .store(user_cap, std::sync::atomic::Ordering::Relaxed);
    }

    /// Check the two pre-authentication login rate-limit buckets.
    ///
    /// Both `login_ip:{addr}` and `login_user:{username}` are consulted.
    /// Each failed attempt ALWAYS consumes a token from the IP bucket
    /// (the username may be unknown or wrong, but the IP is always real).
    /// The user bucket is only consumed when a username is provided.
    ///
    /// Capacities come from the values set via [`set_login_capacities`].
    /// Each bucket refills at `capacity / 60` tokens per second — one full
    /// window per minute.
    ///
    /// Returns [`LoginRateLimitOutcome::Allowed`] when both buckets have tokens.
    pub fn check_login(&self, peer_addr: &str, username: &str) -> LoginRateLimitOutcome {
        let ip_cap = self.login_ip_cap.load(std::sync::atomic::Ordering::Relaxed);
        let user_cap = self
            .login_user_cap
            .load(std::sync::atomic::Ordering::Relaxed);

        // 0-cap means the bucket type is disabled.
        if ip_cap > 0 {
            let ip_key = format!("login_ip:{peer_addr}");
            let ip_rate = (ip_cap as f64) / 60.0;
            if !self.check_login_bucket(&ip_key, ip_cap, ip_rate) {
                return LoginRateLimitOutcome::IpExceeded;
            }
        }

        if user_cap > 0 && !username.is_empty() {
            let user_key = format!("login_user:{username}");
            let user_rate = (user_cap as f64) / 60.0;
            if !self.check_login_bucket(&user_key, user_cap, user_rate) {
                return LoginRateLimitOutcome::UserExceeded;
            }
        }

        LoginRateLimitOutcome::Allowed
    }

    /// Check a login-specific bucket with an explicit refill rate.
    ///
    /// Creates the bucket with the given capacity and rate if it does not exist.
    /// Returns `true` (allowed) or `false` (rate-limited).
    fn check_login_bucket(&self, key: &str, capacity: u64, rate_per_sec: f64) -> bool {
        // Fast path: read-only check.
        {
            let buckets = self.buckets.read().unwrap_or_else(|p| p.into_inner());
            if let Some(bucket) = buckets.get(key) {
                return bucket.try_acquire(1);
            }
        }
        // Slow path: create bucket.
        let mut buckets = self.buckets.write().unwrap_or_else(|p| p.into_inner());
        let bucket = buckets
            .entry(key.to_string())
            .or_insert_with(|| TokenBucket::new(capacity, rate_per_sec));
        bucket.try_acquire(1)
    }

    /// Check rate limit for a request.
    ///
    /// `user_id` = authenticated user.
    /// `org_ids` = user's org memberships (for org-level rate limiting).
    /// `plan_tier` = tier name from `$auth.metadata.plan` (e.g., "free", "pro").
    /// `operation` = endpoint name for cost multiplier lookup.
    /// `quota` = optional tenant/database QPS caps; pass `None` to skip those buckets.
    ///
    /// Check order: user → org → tenant → database. First denial wins.
    pub fn check(
        &self,
        user_id: &str,
        org_ids: &[String],
        plan_tier: Option<&str>,
        operation: &str,
        quota: Option<&QuotaCheckParams>,
    ) -> RateLimitResult {
        if !self.config.enabled {
            return RateLimitResult {
                allowed: true,
                remaining: u64::MAX,
                limit: u64::MAX,
                retry_after_secs: 0,
            };
        }

        let cost = self.config.operation_cost(operation);

        // Resolve the tier (from JWT plan claim or default).
        let (qps, burst) = self.resolve_tier(plan_tier);

        // Check user-level bucket.
        let user_key = format!("user:{user_id}");
        let user_result = self.check_bucket(&user_key, qps, burst, cost);

        if !user_result.allowed {
            debug!(
                user_id = %user_id,
                operation = %operation,
                cost,
                "rate limited (user bucket)"
            );
            return user_result;
        }

        // Check org-level bucket (shared across members).
        for org_id in org_ids {
            let org_key = format!("org:{org_id}");
            // Org gets 10x the user rate (shared budget).
            let org_result = self.check_bucket(&org_key, qps * 10, burst * 10, cost);
            if !org_result.allowed {
                debug!(
                    user_id = %user_id,
                    org_id = %org_id,
                    operation = %operation,
                    "rate limited (org bucket)"
                );
                return org_result;
            }
        }

        // Check tenant-level bucket (if a cap is configured).
        if let Some(q) = quota {
            if q.tenant_max_qps.is_some_and(|v| v > 0) {
                let tenant_qps = q.tenant_max_qps.unwrap_or(0);
                let tenant_key = format!("tenant:{}", q.tenant_id.as_u64());
                let tenant_result = self.check_bucket(&tenant_key, tenant_qps, tenant_qps, cost);
                if !tenant_result.allowed {
                    debug!(
                        tenant_id = q.tenant_id.as_u64(),
                        operation = %operation,
                        "rate limited (tenant bucket)"
                    );
                    return tenant_result;
                }
            }

            // Check database-level bucket (if a cap is configured).
            if q.database_max_qps.is_some_and(|v| v > 0) {
                let db_qps = q.database_max_qps.unwrap_or(0);
                let db_key = format!("database:{}", q.database_id.as_u64());
                let db_result = self.check_bucket(&db_key, db_qps, db_qps, cost);
                if !db_result.allowed {
                    debug!(
                        database_id = q.database_id.as_u64(),
                        operation = %operation,
                        "rate limited (database bucket)"
                    );
                    return db_result;
                }
            }
        }

        user_result
    }

    /// Check with per-API-key limits (independent bucket).
    pub fn check_api_key(
        &self,
        key_id: &str,
        max_qps: u64,
        max_burst: u64,
        operation: &str,
    ) -> RateLimitResult {
        if !self.config.enabled || max_qps == 0 {
            return RateLimitResult {
                allowed: true,
                remaining: u64::MAX,
                limit: u64::MAX,
                retry_after_secs: 0,
            };
        }
        let cost = self.config.operation_cost(operation);
        let key = format!("apikey:{key_id}");
        self.check_bucket(&key, max_qps, max_burst, cost)
    }

    /// Check a single bucket, creating it if it doesn't exist.
    fn check_bucket(&self, key: &str, qps: u64, burst: u64, cost: u64) -> RateLimitResult {
        // Fast path: read-only check.
        {
            let buckets = self.buckets.read().unwrap_or_else(|p| p.into_inner());
            if let Some(bucket) = buckets.get(key) {
                let allowed = bucket.try_acquire(cost);
                return RateLimitResult {
                    allowed,
                    remaining: bucket.available(),
                    limit: bucket.capacity(),
                    retry_after_secs: if allowed {
                        0
                    } else {
                        (bucket.retry_after_ms() / 1000).max(1)
                    },
                };
            }
        }

        // Slow path: create bucket.
        let mut buckets = self.buckets.write().unwrap_or_else(|p| p.into_inner());
        let bucket = buckets
            .entry(key.to_string())
            .or_insert_with(|| TokenBucket::new(burst, qps as f64));

        let allowed = bucket.try_acquire(cost);
        RateLimitResult {
            allowed,
            remaining: bucket.available(),
            limit: bucket.capacity(),
            retry_after_secs: if allowed {
                0
            } else {
                (bucket.retry_after_ms() / 1000).max(1)
            },
        }
    }

    /// Resolve rate limit tier from plan name.
    fn resolve_tier(&self, plan_tier: Option<&str>) -> (u64, u64) {
        if let Some(tier_name) = plan_tier
            && let Some(tier) = self.config.tier(tier_name)
        {
            return (tier.qps, tier.burst);
        }
        (self.config.default_qps, self.config.default_burst)
    }

    /// Build HTTP response headers for rate limit info.
    pub fn response_headers(result: &RateLimitResult) -> Vec<(String, String)> {
        vec![
            ("X-RateLimit-Limit".into(), result.limit.to_string()),
            ("X-RateLimit-Remaining".into(), result.remaining.to_string()),
            (
                "X-RateLimit-Reset".into(),
                result.retry_after_secs.to_string(),
            ),
        ]
    }

    /// Build Retry-After header value (seconds).
    pub fn retry_after_header(result: &RateLimitResult) -> Option<(String, String)> {
        if result.allowed {
            None
        } else {
            Some(("Retry-After".into(), result.retry_after_secs.to_string()))
        }
    }

    /// Record a rate limit rejection and return the total count.
    /// Exposed as `nodedb_rate_limit_rejected_total` in Prometheus metrics.
    pub fn record_rejection(&self) -> u64 {
        self.rejections_total
            .fetch_add(1, std::sync::atomic::Ordering::Relaxed)
            + 1
    }

    /// Get total rejection count for Prometheus export.
    pub fn rejections_total(&self) -> u64 {
        self.rejections_total
            .load(std::sync::atomic::Ordering::Relaxed)
    }

    /// Whether rate limiting is enabled.
    pub fn is_enabled(&self) -> bool {
        self.config.enabled
    }

    /// Number of active buckets (for metrics).
    pub fn active_buckets(&self) -> usize {
        self.buckets.read().unwrap_or_else(|p| p.into_inner()).len()
    }

    /// Get the config for inspection.
    pub fn config(&self) -> &RateLimitConfig {
        &self.config
    }
}

impl Default for RateLimiter {
    fn default() -> Self {
        Self::new(RateLimitConfig::default())
    }
}

impl std::fmt::Debug for RateLimiter {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("RateLimiter")
            .field(
                "login_ip_cap",
                &self.login_ip_cap.load(std::sync::atomic::Ordering::Relaxed),
            )
            .field(
                "login_user_cap",
                &self
                    .login_user_cap
                    .load(std::sync::atomic::Ordering::Relaxed),
            )
            .finish_non_exhaustive()
    }
}

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

    fn enabled_config() -> RateLimitConfig {
        use crate::control::security::ratelimit::config::RateLimitTier;
        let mut config = RateLimitConfig {
            enabled: true,
            default_qps: 10,
            default_burst: 20,
            ..Default::default()
        };
        config.tiers.insert(
            "pro".into(),
            RateLimitTier {
                qps: 5000,
                burst: 10000,
            },
        );
        config
    }

    #[test]
    fn disabled_allows_all() {
        let limiter = RateLimiter::new(RateLimitConfig::default());
        let result = limiter.check("u1", &[], None, "point_get", None);
        assert!(result.allowed);
    }

    #[test]
    fn basic_rate_limiting() {
        let limiter = RateLimiter::new(enabled_config());

        // Burst of 20, cost 1 each.
        for _ in 0..20 {
            let r = limiter.check("u1", &[], None, "point_get", None);
            assert!(r.allowed);
        }
        // 21st request should be rejected.
        let r = limiter.check("u1", &[], None, "point_get", None);
        assert!(!r.allowed);
        assert!(r.retry_after_secs > 0);
    }

    #[test]
    fn cost_multiplier_drains_faster() {
        let limiter = RateLimiter::new(enabled_config());

        // vector_search costs 20 tokens. Burst is 20. First request OK.
        let r = limiter.check("u1", &[], None, "vector_search", None);
        assert!(r.allowed);
        // Second should fail (20 tokens consumed, 0 remaining).
        let r = limiter.check("u1", &[], None, "vector_search", None);
        assert!(!r.allowed);
    }

    #[test]
    fn tier_resolution() {
        let limiter = RateLimiter::new(enabled_config());

        // Pro tier: 5000 QPS, 10000 burst.
        for _ in 0..100 {
            let r = limiter.check("u1", &[], Some("pro"), "point_get", None);
            assert!(r.allowed);
        }
    }

    #[test]
    fn per_user_isolation() {
        let limiter = RateLimiter::new(enabled_config());

        // Exhaust u1's bucket.
        for _ in 0..20 {
            limiter.check("u1", &[], None, "point_get", None);
        }
        let r = limiter.check("u1", &[], None, "point_get", None);
        assert!(!r.allowed);

        // u2 should still have tokens.
        let r = limiter.check("u2", &[], None, "point_get", None);
        assert!(r.allowed);
    }

    #[test]
    fn response_headers() {
        let result = RateLimitResult {
            allowed: true,
            remaining: 50,
            limit: 100,
            retry_after_secs: 0,
        };
        let headers = RateLimiter::response_headers(&result);
        assert_eq!(headers.len(), 3);
        assert_eq!(headers[0].0, "X-RateLimit-Limit");
        assert_eq!(headers[0].1, "100");
    }

    // ── Login rate-limit tests ───────────────────────────────────────

    fn login_limiter(ip_cap: u64, user_cap: u64) -> RateLimiter {
        let limiter = RateLimiter::new(RateLimitConfig::default());
        limiter.set_login_capacities(ip_cap, user_cap);
        limiter
    }

    #[test]
    fn login_rate_limit_ip() {
        let limiter = login_limiter(30, 10);

        // 30 attempts from one IP — all allowed.
        for i in 0..30 {
            let outcome = limiter.check_login("10.0.0.1", &format!("user_{i}"));
            assert!(
                matches!(outcome, LoginRateLimitOutcome::Allowed),
                "attempt {i} should be allowed"
            );
        }
        // 31st attempt — IP bucket exhausted.
        let outcome = limiter.check_login("10.0.0.1", "user_overflow");
        assert!(
            matches!(outcome, LoginRateLimitOutcome::IpExceeded),
            "31st attempt from same IP must be rate-limited"
        );

        // Different IP is unaffected.
        let outcome = limiter.check_login("10.0.0.2", "user_other");
        assert!(
            matches!(outcome, LoginRateLimitOutcome::Allowed),
            "different IP must still be allowed"
        );
    }

    #[test]
    fn login_rate_limit_user() {
        let limiter = login_limiter(30, 10);

        // 10 attempts for the same username from different IPs — all allowed.
        for i in 0..10 {
            let outcome = limiter.check_login(&format!("10.0.0.{i}"), "victim");
            assert!(
                matches!(outcome, LoginRateLimitOutcome::Allowed),
                "attempt {i} should be allowed"
            );
        }
        // 11th attempt — user bucket exhausted.
        let outcome = limiter.check_login("10.0.0.200", "victim");
        assert!(
            matches!(outcome, LoginRateLimitOutcome::UserExceeded),
            "11th attempt for same user must be rate-limited"
        );

        // Different username is unaffected.
        let outcome = limiter.check_login("10.0.0.200", "other_user");
        assert!(
            matches!(outcome, LoginRateLimitOutcome::Allowed),
            "different username must still be allowed"
        );
    }

    #[test]
    fn login_rate_limit_window() {
        // Use a small capacity (2) so the window reset is observable
        // without sleeping 60 seconds.  The bucket refills at 2/60 tokens/s.
        // We exhaust it, then verify the bucket is a real TokenBucket that
        // will refill given elapsed time — confirm via `available()`.
        let limiter = login_limiter(2, 100);

        assert!(matches!(
            limiter.check_login("192.0.2.1", "u"),
            LoginRateLimitOutcome::Allowed
        ));
        assert!(matches!(
            limiter.check_login("192.0.2.1", "u"),
            LoginRateLimitOutcome::Allowed
        ));
        // Third attempt — exhausted.
        assert!(matches!(
            limiter.check_login("192.0.2.1", "u"),
            LoginRateLimitOutcome::IpExceeded
        ));

        // After the bucket is exhausted the `available()` is 0.
        {
            let buckets = limiter.buckets.read().unwrap_or_else(|p| p.into_inner());
            let bucket = buckets
                .get("login_ip:192.0.2.1")
                .expect("bucket must exist");
            assert_eq!(
                bucket.available(),
                0,
                "bucket must be empty after exhaustion"
            );
        }
    }

    #[test]
    fn login_rate_limit_audit() {
        use crate::control::security::audit::emitter::test_helpers::CapturingEmitter;
        use crate::control::security::audit::emitter::{AuditEmitContext, AuditEmitter};
        use crate::control::security::audit::event::AuditEvent;

        let emitter = CapturingEmitter::new();
        emitter.emit(
            AuditEvent::LoginRateLimited,
            "login_rate_limit",
            "ip=10.0.0.1 user=alice",
            AuditEmitContext::new(None, "", "alice"),
        );

        let recorded = emitter.recorded();
        assert_eq!(recorded.len(), 1);
        assert_eq!(recorded[0].0, AuditEvent::LoginRateLimited);
        assert!(recorded[0].2.contains("alice"));
    }

    #[test]
    fn login_rate_limit_constant_time() {
        use std::time::Instant;

        // Simulate the constant-time floor by measuring that an immediate
        // rejection (rate-limited before any Argon2) cannot be distinguished
        // from a real Argon2 rejection by timing alone.  We can't run actual
        // Argon2 here (test suite must be fast), so we verify the *floor
        // constant* is well-defined (non-zero) and that the enforcement
        // mechanism is present in the public API surface.
        //
        // The real enforcement is in `session_auth` (production code).
        // Here we only verify the rate-limit decision itself is fast
        // (sub-millisecond) so the test detects accidental blocking in the
        // decision path — the caller adds the floor separately.
        let limiter = login_limiter(5, 5);
        let start = Instant::now();
        for i in 0..10 {
            let _ = limiter.check_login("10.1.2.3", &format!("user{i}"));
        }
        let elapsed = start.elapsed();
        // 10 check_login calls must complete in under 10ms (no blocking).
        assert!(
            elapsed.as_millis() < 10,
            "check_login must be non-blocking; took {elapsed:?}"
        );
    }

    // ── Tenant and database bucket tests ────────────────────────────────────

    fn db_id() -> DatabaseId {
        DatabaseId::DEFAULT
    }

    fn t_id(n: u64) -> TenantId {
        TenantId::new(n)
    }

    #[test]
    fn database_cap_deny_while_tenant_has_headroom() {
        let limiter = RateLimiter::new(enabled_config());

        // Database cap: 5 (burst = 5). Tenant cap: 1000 (generous).
        let quota = QuotaCheckParams {
            tenant_max_qps: Some(1000),
            database_max_qps: Some(5),
            tenant_id: t_id(1),
            database_id: db_id(),
        };

        // Consume the database bucket.
        for _ in 0..5 {
            let r = limiter.check("u1", &[], None, "point_get", Some(&quota));
            assert!(r.allowed, "first 5 should be allowed under database cap");
        }

        // 6th request: database bucket exhausted even though tenant bucket is fine.
        let r = limiter.check("u1", &[], None, "point_get", Some(&quota));
        assert!(
            !r.allowed,
            "database bucket exhausted — request must be denied"
        );
    }

    #[test]
    fn tenant_cap_deny_while_database_has_headroom() {
        let limiter = RateLimiter::new(enabled_config());

        // Tenant cap: 3. Database cap: 1000 (generous).
        let quota = QuotaCheckParams {
            tenant_max_qps: Some(3),
            database_max_qps: Some(1000),
            tenant_id: t_id(2),
            database_id: db_id(),
        };

        // Consume the tenant bucket.
        for _ in 0..3 {
            let r = limiter.check("u2", &[], None, "point_get", Some(&quota));
            assert!(r.allowed, "first 3 should be allowed under tenant cap");
        }

        // 4th request: tenant bucket exhausted.
        let r = limiter.check("u2", &[], None, "point_get", Some(&quota));
        assert!(
            !r.allowed,
            "tenant bucket exhausted — request must be denied"
        );
    }

    #[test]
    fn when_both_would_deny_tenant_wins_over_database() {
        // Both caps = 1.  The user bucket has burst 20, so it won't deny.
        // The tenant bucket is checked before the database bucket, so
        // whichever fires first is the tenant bucket.
        let limiter = RateLimiter::new(enabled_config());

        let quota = QuotaCheckParams {
            tenant_max_qps: Some(1),
            database_max_qps: Some(1),
            tenant_id: t_id(3),
            database_id: db_id(),
        };

        // First request — allowed, drains both caps.
        let r = limiter.check("u3", &[], None, "point_get", Some(&quota));
        assert!(r.allowed, "first request should be allowed");

        // Second request — tenant bucket fires first (checked before database).
        let r2 = limiter.check("u3", &[], None, "point_get", Some(&quota));
        assert!(!r2.allowed, "second request must be denied");
        // We can't directly observe *which* bucket denied without instrumenting
        // the limiter further, but we can assert that a new quota with only a
        // database cap (no tenant cap) is ALSO denied at this point — confirming
        // the database bucket was also consumed.
        let quota_db_only = QuotaCheckParams {
            tenant_max_qps: None,
            database_max_qps: Some(1),
            tenant_id: t_id(3),
            database_id: db_id(),
        };
        let r3 = limiter.check("u3", &[], None, "point_get", Some(&quota_db_only));
        assert!(!r3.allowed, "database bucket should also be exhausted");
    }

    #[test]
    fn no_quota_params_skips_tenant_and_database_buckets() {
        let limiter = RateLimiter::new(enabled_config());
        // With no quota params, only user/org buckets apply.
        // Burst = 20, so 20 requests allowed.
        for _ in 0..20 {
            let r = limiter.check("u4", &[], None, "point_get", None);
            assert!(r.allowed);
        }
        // 21st exceeds user burst (not tenant/db).
        let r = limiter.check("u4", &[], None, "point_get", None);
        assert!(!r.allowed);
    }
}