Design an API Rate Limiter

local key = KEYS[1]
local capacity = tonumber(ARGV[1])
local refill_rate = tonumber(ARGV[2])       -- tokens added per second (must be > 0)
local requested_cost = tonumber(ARGV[3]) or 1

-- For simplicity, current time is supplied by the caller.
-- In multi-gateway deployments where clock skew matters,
-- use Redis TIME or another trusted synchronized clock source.
local now = tonumber(ARGV[4])               -- epoch timestamp in seconds

-- Defensive configuration and parameter validation
if capacity == nil or capacity <= 0 then
    return redis.error_reply("invalid rate limit configuration: capacity must be > 0")
end

if refill_rate == nil or refill_rate <= 0 then
    return redis.error_reply("invalid rate limit configuration: refill_rate must be > 0")
end

if requested_cost == nil or requested_cost <= 0 then
    return redis.error_reply("invalid request cost: requested_cost must be > 0")
end

if now == nil or now <= 0 then
    return redis.error_reply("invalid timestamp")
end

-- Invariant: requested cost cannot exceed maximum bucket capacity
if requested_cost > capacity then
    return {0, 0, -1}  -- [0 = rejected, remaining = 0, retry_after = -1 (permanently impossible under policy)]
end

local data = redis.call('HMGET', key, 'tokens', 'last_refill')
local tokens = tonumber(data[1])
local last_refill = tonumber(data[2])

if tokens == nil or last_refill == nil then
    -- Initialize fresh bucket to full capacity
    tokens = capacity
    last_refill = now
else
    -- Refill tokens based on elapsed wall time
    local elapsed = math.max(0, now - last_refill)
    tokens = math.min(capacity, tokens + elapsed * refill_rate)
    last_refill = now
end

if tokens >= requested_cost then
    tokens = tokens - requested_cost
    redis.call('HMSET', key, 'tokens', tokens, 'last_refill', last_refill)
    -- Idle-state cleanup TTL: time required to refill from empty to full plus a safety margin
    local refill_duration = math.ceil(capacity / refill_rate)
    local ttl = math.max(refill_duration * 2, 60)
    redis.call('EXPIRE', key, ttl)
    return {1, math.floor(tokens), 0}  -- [1 = allowed, remaining_tokens, retry_after = 0]
else
    redis.call('HMSET', key, 'tokens', tokens, 'last_refill', last_refill)
    local tokens_needed = requested_cost - tokens
    local retry_after = math.ceil(tokens_needed / refill_rate)
    return {0, math.floor(tokens), retry_after}  -- [0 = rejected, remaining_tokens, retry_after_seconds]
end

local key = KEYS[1]
-- For simplicity, current time is supplied by the caller in milliseconds.
-- In multi-gateway deployments where clock skew matters, use a synchronized clock source.
local now = tonumber(ARGV[1])          -- current timestamp in milliseconds
local window = tonumber(ARGV[2])       -- window duration in milliseconds
local limit = tonumber(ARGV[3])        -- maximum allowed requests in window (unit-cost requests)
local request_id = ARGV[4]             -- unique request identifier

-- Defensive argument validation
if now == nil or window == nil or limit == nil or window <= 0 or limit <= 0 or request_id == nil then
    return redis.error_reply("invalid sliding window log arguments")
end

-- Active rolling window interval is defined as (now - window, now].
-- Events with timestamp <= cutoff are strictly expired.
local cutoff = now - window

-- 1. Remove all entries older than or equal to the rolling window cutoff
redis.call('ZREMRANGEBYSCORE', key, 0, cutoff)

-- 2. Count currently admitted requests surviving in the active (now - window, now] interval
local count = redis.call('ZCARD', key)

-- 3. Check capacity before admitting (unit-cost enforcement)
if count >= limit then
    -- Find oldest surviving admitted request to calculate precise Retry-After
    local oldest = redis.call('ZRANGE', key, 0, 0, 'WITHSCORES')
    local oldest_ts = tonumber(oldest[2])
    local retry_after_ms = math.max(1, (oldest_ts + window) - now)
    local retry_after_sec = math.ceil(retry_after_ms / 1000)
    return {0, count, retry_after_sec}  -- [0 = rejected, current_count, retry_after_seconds]
else
    -- 4. Admitted: add exactly one unique member with score = now
    local member = now .. ":" .. request_id
    redis.call('ZADD', key, now, member)
    local ttl_sec = math.max(math.ceil((window / 1000) * 2), 60)
    redis.call('EXPIRE', key, ttl_sec)
    return {1, limit - count - 1, 0}    -- [1 = allowed, remaining_quota, retry_after = 0]
end

local current_key = KEYS[1]
local previous_key = KEYS[2]
local limit = tonumber(ARGV[1])
local window_size = tonumber(ARGV[2])    -- window size in seconds (e.g., 60)
-- For simplicity, current elapsed seconds supplied by caller.
local current_second = tonumber(ARGV[3])  -- seconds elapsed in current window (0 to window_size - 1)
local requested_cost = tonumber(ARGV[4]) or 1

-- Defensive parameter validation
if limit == nil or limit <= 0 or window_size == nil or window_size <= 0 or current_second == nil or requested_cost == nil or requested_cost <= 0 then
    return redis.error_reply("invalid sliding window counter arguments")
end

local prev_count = tonumber(redis.call('GET', previous_key)) or 0
local curr_count = tonumber(redis.call('GET', current_key)) or 0

-- Time-weighted linear approximation of requests admitted in the rolling window
local weight = (window_size - current_second) / window_size
local estimated_before = prev_count * weight + curr_count

-- Check if admitting this request exceeds the configured limit
if estimated_before + requested_cost > limit then
    -- Retry-After is a conservative approximation: seconds until current window rolls over
    -- (Actual capacity continuously decays as the previous window weight decreases)
    local retry_after = math.max(1, window_size - current_second)
    return {0, math.floor(estimated_before), retry_after}  -- [0 = rejected, estimated_count, retry_after]
else
    local new_count = redis.call('INCRBY', current_key, requested_cost)
    if new_count == requested_cost then
        redis.call('EXPIRE', current_key, window_size * 2)
    end
    -- Estimated remaining capacity (conservative integer floor approximation to avoid overstating quota)
    local estimated_remaining = math.max(0, math.floor(limit - (estimated_before + requested_cost)))
    return {1, estimated_remaining, 0}  -- [1 = allowed, estimated_remaining, 0]
end

# Kafka Event Bus Topology for Rules Distribution and Violation Auditing

topics:
  rate-rules-updates:
    partitions: 8 # low volume for configuration updates
    partition_key: rule_id
    retention: 7 days
    replication_factor: 3
    min_insync_replicas: 2
    producer: "Rules Configuration Service (triggers on admin CRUD operations)"
    consumer_group: "gateway-rule-sync"
    consumer_behavior:
      - "Each API Gateway or sidecar proxy refreshes its local in-memory rules cache"
      - "Message payload contains monotonically increasing rule versions to ignore stale updates"
      - "Lag alerting triggers if consumer lag exceeds 30 seconds (SLO breach)"

  rate-limit-violations:
    partitions: 64
    partition_key: client_id # preserves per-client ordering for audit trails
    retention: 90 days # compliance and abuse investigations
    replication_factor: 3
    min_insync_replicas: 2
    producer: "Rate Limiter Middleware (asynchronous fire-and-forget on HTTP 429)"
    event_schema:
      event_id: "uuid"
      client_id: "string"
      endpoint: "string"
      tier: "string"
      limit: "integer"
      window_seconds: "integer"
      retry_after_seconds: "integer"
      timestamp: "epoch_seconds"
    consumer_groups:
      abuse-detector: "Flags clients exceeding rate limits repeatedly for automated key suspension or WAF IP blocks"
      analytics-sink: "Streams violation metrics to ClickHouse dashboards for rate limit visibility and capacity planning"
    dead_letter_queue: "rate-limit-violations-dlq (routes events after 3 failed processing attempts)"

execution_paths:
  synchronous_path: "Client request -> extract client identifier -> run Lua check in Redis -> return 200 or 429 (< 1ms)"
  asynchronous_path: "Rule configuration updates propagate via Kafka, while HTTP 429 audit events never block request traffic"