delay-queue-transit.lua

redis.replicate_commands();

local delayBucket = "dq_delay_bucket";
local readyQueuePrefix = "dq_ready_";

while true do
    local first = redis.call("ZRANGE", delayBucket, 0, 0, "WITHSCORES");
    local currTime = redis.call("TIME");

    if first[2] == nil then
        break;
    end

    if currTime[1] < first[2] then 
        break;
    end

    local jobID = first[1];
    local jobMeta = redis.call("GET", jobID);
    if jobMeta == nil then
        redis.call("ZREM", delayBucket, jobID);
    end

    local metaJson = cmsgpack.unpack(jobMeta);

    redis.call("RPUSH", readyQueuePrefix .. metaJson.topic, jobID);
    redis.call("ZREM", delayBucket, jobID);
end