-- Fairness helpers shared by all grant scripts. This partial is concatenated in front of each
-- grant script by RuleExecutor.registerScripts(), so every grant decision, waiter registration
-- and waiter pruning happens inside a single atomic script execution.
--
-- Waiters are kept in a per-rule-key sorted set (KEYS[2] of every grant script):
--   member = '<waiterId>:<maxWaitMs>', score = Redis server time in ms at registration.
-- The deadline of a waiter is derivable from the entry alone (score + maxWaitMs), so entries
-- left behind by crashed nodes are lazily pruned by any later script call — the queue can never
-- be blocked for longer than the dead waiter's own maxWait.
--
-- The grant rule evaluated by every script is:
--   grant <=> liveWaitersAheadOfCaller < freeSlots
-- A fresh (non-waiting) caller counts every live waiter as "ahead", so it can never steal
-- capacity from the queue; a registered waiter only counts waiters older than itself, so
-- several waiters can be granted in one burst (no head-of-line blocking).

-- Scripts below call TIME (non-deterministic) before writing; switch to effect replication on
-- Redis < 5 where it is not the default. No-op on modern Redis.
if redis.replicate_commands then
    redis.replicate_commands()
end

-- Current Redis server time in ms. Using the Redis clock (instead of node clocks passed in via
-- ARGV) keeps waiter ordering and window math skew-free across engine nodes.
local function now_ms()
    local t = redis.call('TIME')
    return tonumber(t[1]) * 1000 + math.floor(tonumber(t[2]) / 1000)
end

-- Registration score: milliseconds with the sub-millisecond remainder kept as a fraction
-- (Redis TIME has microsecond resolution). Two waiters registering within the same
-- millisecond would otherwise share a score and be ordered by member string instead of
-- by arrival.
local function now_score()
    local t = redis.call('TIME')
    return tonumber(t[1]) * 1000 + tonumber(t[2]) / 1000
end

-- Number of live waiters strictly ahead of waiterId ('' for fresh callers, which ranks behind
-- everyone). Prunes expired waiters (deadline passed) while scanning.
local function waiters_ahead(waitersKey, waiterId, now)
    if redis.call('ZCARD', waitersKey) == 0 then
        return 0
    end
    local members = redis.call('ZRANGE', waitersKey, 0, -1, 'WITHSCORES')
    local ahead = 0
    for i = 1, #members, 2 do
        local member = members[i]
        local score = tonumber(members[i + 1])
        local id, maxWait = string.match(member, '^(.+):(%d+)$')
        if id == nil or score + tonumber(maxWait) <= now then
            redis.call('ZREM', waitersKey, member)
        elseif id == waiterId then
            return ahead
        else
            ahead = ahead + 1
        end
    end
    return ahead
end

-- Register the caller as a waiter (called on denial when the caller is willing to wait).
-- ZADD NX keeps the original score on repeated attempts, so the caller's queue position is
-- stable across retries.
local function register_waiter(waitersKey, waiterId, maxWaitMs)
    if waiterId == '' then
        return
    end
    redis.call('ZADD', waitersKey, 'NX', now_score(), waiterId .. ':' .. maxWaitMs)
    -- Make sure the key outlives the longest pending deadline, so abandoned queues are
    -- garbage-collected by Redis itself.
    local ttlNeeded = maxWaitMs + 5000
    local ttl = redis.call('PTTL', waitersKey)
    if ttl < ttlNeeded then
        redis.call('PEXPIRE', waitersKey, ttlNeeded)
    end
end

-- Remove the caller from the waiter queue (called when its grant succeeds).
local function deregister_waiter(waitersKey, waiterId, maxWaitMs)
    if waiterId ~= '' then
        redis.call('ZREM', waitersKey, waiterId .. ':' .. maxWaitMs)
    end
end