-- Concurrency quota grant script (the fairness partial is prepended by RuleExecutor).
-- Tracks each active grant as a member of a sorted set, scored by its expiration
-- timestamp. Grants whose owner crashes before dismiss are reclaimed automatically
-- by ZREMRANGEBYSCORE on the next call, so the limit cannot be permanently inflated
-- by orphaned counters (see issue #2999).
--
-- KEYS[1]: sorted set key (active grants)
-- KEYS[2]: waiters key (sorted set of waiters, see partials/fairness.lua)
-- ARGV[1]: limit
-- ARGV[2]: quotaId (member added to the set when granted)
-- ARGV[3]: grant ttl (ms) — how long a grant is considered alive without a dismiss
-- ARGV[4]: waiterId ('' for a fresh, non-waiting caller)
-- ARGV[5]: maxWait (ms) of the caller
-- ARGV[6]: mayWait (1 = register as waiter on denial, 0 = never)
-- Returns: {granted (0/1), currentCount, retryIn (ms, -1 = no hint)}

local limit = tonumber(ARGV[1])
local ttl = tonumber(ARGV[3])
local now = now_ms()

redis.call('ZREMRANGEBYSCORE', KEYS[1], '-inf', now)

local currentCount = redis.call('ZCARD', KEYS[1])
local free = limit - currentCount
local ahead = waiters_ahead(KEYS[2], ARGV[4], now)

if ahead < free then
    redis.call('ZADD', KEYS[1], now + ttl, ARGV[2])
    redis.call('PEXPIRE', KEYS[1], ttl)
    deregister_waiter(KEYS[2], ARGV[4], ARGV[5])
    return { 1, currentCount + 1, -1 }
end

if tonumber(ARGV[6]) == 1 then
    register_waiter(KEYS[2], ARGV[4], ARGV[5])
end

-- Grant scores are expiry timestamps, so the entry at the caller's index is an upper bound on
-- when a slot frees — a dismiss PUBLISH normally wakes the waiter much earlier.
local retryIn = -1
if currentCount > 0 then
    local idx = math.min(math.max(ahead + 1 - free, 1) - 1, currentCount - 1)
    local entry = redis.call('ZRANGE', KEYS[1], idx, idx, 'WITHSCORES')
    retryIn = math.max(math.ceil(tonumber(entry[2]) - now), 1)
end

return { 0, currentCount, retryIn }