Redis Scheduled Jobs

zaddhset.lua

-- zaddhset sortedsetkey score hsetkey fieldname fieldvalue [fieldname fieldvalue ...]
-- sortedsetkey = KEYS[1]
-- score = KEYS[2]
-- hsetkey = KEYS[3]
-- fields = ARGV
if #KEYS ~= 3 or #ARGV < 2 or #ARGV % 2 ~= 0 then
  return {err="ERR wrong number of arguments for 'zaddhset' command"}
end

if redis.call('EXISTS', KEYS[3]) == 1 then return nil end
redis.call('ZADD', KEYS[1], KEYS[2], KEYS[3])
redis.call('HSET', KEYS[3], unpack(ARGV))
return 'OK'

zrangehgetallxadd.lua

-- zrangehgetallxadd sortedsetkey streamkey minscore maxscore
-- sortedsetkey = KEYS[1]
-- streamkey = KEYS[2]
-- minscore = ARGV[1]
-- maxscore = ARGV[2]
if #KEYS ~= 2 or #ARGV ~= 2 then
  return {err="ERR wrong number of arguments for 'zrangehgetallxadd' command"}
end

local hashkeys = redis.call('ZRANGEBYSCORE', KEYS[1], ARGV[1], ARGV[2])
if not hashkeys or #hashkeys == 0 then return 0 end

for i = 1, #hashkeys do
  local fields = redis.call('HGETALL', hashkeys[i])
  if #fields ~= 0 then redis.call('XADD', KEYS[2], '*', unpack(fields)) end
end

return redis.call('ZREM', KEYS[1], unpack(hashkeys))

1. Call zaddhset to queue a job, deduplication happens by hsetkey passed in.
2. Call zrangehgetallxadd peridically to move the jobs into a redis stream