local records_key = KEYS[1]
local owners_key = KEYS[2]
local worker_owned_key = KEYS[3]

local worker_id = ARGV[1]
local limit = tonumber(ARGV[2])

local function reply(status, proxies)
    if not proxies or #proxies == 0 then
        return '{"status":' .. cjson.encode(status) .. ',"proxies":[]}'
    end
    return cjson.encode({status = status, proxies = proxies})
end

local function now_ms()
    local value = redis.call('TIME')
    return tonumber(value[1]) * 1000 + math.floor(tonumber(value[2]) / 1000)
end

local function decode_table(value)
    if not value then
        return nil
    end
    local ok, decoded = pcall(cjson.decode, value)
    if not ok or type(decoded) ~= 'table' then
        return nil
    end
    return decoded
end

local function valid_assignment(value, now)
    return value and value.version == 1 and type(value.proxyId) == 'string' and value.proxyId ~= '' and
        type(value.workerId) == 'string' and value.workerId == worker_id and type(value.epoch) == 'number' and
        value.epoch > 0 and type(value.expiresAtMs) == 'number' and value.expiresAtMs > now and
        type(value.draining) == 'boolean' and not value.draining and
        type(value.workerIndexKey) == 'string' and value.workerIndexKey == worker_owned_key
end

local function valid_record(value, proxy_id, now)
    return value and value.version == 1 and value.id == proxy_id and value.state == 'AVAILABLE' and
        value.ownerWorkerId == worker_id and type(value.scheme) == 'string' and
        (value.scheme == 'http' or value.scheme == 'https' or value.scheme == 'socks5') and
        type(value.host) == 'string' and value.host ~= '' and type(value.port) == 'number' and
        value.port > 0 and value.port <= 65535 and type(value.sourceUpstream) == 'string' and
        value.sourceUpstream ~= '' and type(value.expiresAtMs) == 'number' and value.expiresAtMs > now and
        type(value.usableUntilMs) == 'number' and value.usableUntilMs > now and
        type(value.maxConcurrency) == 'number' and value.maxConcurrency > 0
end

if type(worker_id) ~= 'string' or worker_id == '' or not limit or limit <= 0 then
    return reply('invalid')
end

if redis.call('ZCARD', worker_owned_key) > limit then
    return reply('unavailable')
end

local now = now_ms()
local ids = redis.call('ZRANGEBYSCORE', worker_owned_key, now + 1, '+inf', 'LIMIT', 0, limit)
local proxies = cjson.decode('[]')
for _, proxy_id in ipairs(ids) do
    local assignment = decode_table(redis.call('HGET', owners_key, proxy_id))
    local raw_record = redis.call('HGET', records_key, proxy_id)
    local record = decode_table(raw_record)
    if valid_assignment(assignment, now) and valid_record(record, proxy_id, now) then
        proxies[#proxies + 1] = {
            record = raw_record,
            ownershipEpoch = tostring(assignment.epoch),
            leaseExpiresAtMs = assignment.expiresAtMs,
        }
    else
        redis.call('ZREM', worker_owned_key, proxy_id)
    end
end
return reply('ok', proxies)