mhenrixon/sidekiq-unique-jobs

-------- BEGIN keys ---------
local digest    = KEYS[1]
local queued    = KEYS[2]
local primed    = KEYS[3]
local locked    = KEYS[4]
local info      = KEYS[5]
local changelog = KEYS[6]
local digests   = KEYS[7]
-------- END keys ---------


-------- BEGIN lock arguments ---------
local job_id     = ARGV[1]      -- The job_id that was previously primed
local pttl       = tonumber(ARGV[2])
local lock_type  = ARGV[3]
local limit      = tonumber(ARGV[4])
local lock_score = ARGV[5]
-------- END lock arguments -----------


--------  BEGIN injected arguments --------
local current_time = tonumber(ARGV[6])
local debug_lua    = tostring(ARGV[7]) == "1"
local max_history  = tonumber(ARGV[8])
local script_name  = tostring(ARGV[9]) .. ".lua"
---------  END injected arguments ---------


--------  BEGIN Variables --------
local queued_count = redis.call("LLEN", queued)
local locked_count = redis.call("HLEN", locked)
local within_limit = limit > locked_count
local limit_exceeded = not within_limit
--------   END Variables  --------


--------  BEGIN local functions --------
<%= include_partial "shared/_common.lua" %>
----------  END local functions ----------


--------  BEGIN queue.lua --------
log_debug("BEGIN queue with key:", digest, "for job:", job_id)

if redis.call("HEXISTS", locked, job_id) == 1 then
  log_debug("HEXISTS", locked, job_id, "== 1")
  log("Duplicate")
  return job_id
end

local prev_jid = redis.call("GET", digest)
log_debug("job_id:", job_id, "prev_jid:", prev_jid)
if not prev_jid or prev_jid == false then
  log_debug("SET", digest, job_id)
  redis.call("SET", digest, job_id)
elseif prev_jid == job_id then
  log_debug(digest, "already queued with job_id:", job_id)
  log("Duplicate")
  return job_id
else
  -- TODO: Consider constraining the total count of both locked and queued?
  if within_limit and queued_count < limit then
    log_debug("Within limit:", digest, "(",  locked_count, "of", limit, ")", "queued (", queued_count, "of", limit, ")")
    log_debug("SET", digest, job_id, "(was", prev_jid, ")")
    redis.call("SET", digest, job_id)
  else
    log_debug("Limit exceeded:", digest, "(",  locked_count, "of", limit, ")")
    log("Limit exceeded", prev_jid)
    return prev_jid
  end
end

log_debug("LPUSH", queued, job_id)
redis.call("LPUSH", queued, job_id)

-- The Sidekiq client should only set pttl for until_expired
-- The Sidekiq server should set pttl for all other jobs
if pttl and pttl > 0 then
  log_debug("PEXPIRE", digest, pttl)
  redis.call("PEXPIRE", digest, pttl)
  log_debug("PEXPIRE", queued, pttl)
  redis.call("PEXPIRE", queued, pttl)
end

log("Queued")
log_debug("END queue with key:", digest, "for job:", job_id)
return job_id
--------  END queue.lua --------