-- argv: ns, groupId, dataJson, maxAttempts, orderMs, delayUntil, jobId, keepCompleted, orderingDelayMs, clientTimestamp
local ns = ARGV[1]
local groupId = ARGV[2]
local data = ARGV[3]
local maxAttempts = tonumber(ARGV[4])
local orderMs = tonumber(ARGV[5])
local delayUntil = tonumber(ARGV[6])
local jobId = ARGV[7]
local keepCompleted = tonumber(ARGV[8]) or 0
local orderingDelayMs = tonumber(ARGV[9]) or 0
local clientTimestamp = tonumber(ARGV[10])

local readyKey = ns .. ":ready"
local delayedKey = ns .. ":delayed"
local jobKey = ns .. ":job:" .. jobId
local groupsKey = ns .. ":groups"

-- Idempotence: ensure unique jobId per queue namespace with stale-key recovery
local uniqueKey = ns .. ":unique:" .. jobId
local uniqueSet = redis.call("SET", uniqueKey, jobId, "NX")
if not uniqueSet then
  -- Duplicate detected. Check for stale unique mapping
  local exists = redis.call("EXISTS", jobKey)
  if exists == 0 then
    -- Job doesn't exist but unique key does (stale), clean up and proceed
    redis.call("DEL", uniqueKey)
    redis.call("SET", uniqueKey, jobId)
  else
    -- Job exists, check its status and location
    local gid = redis.call("HGET", jobKey, "groupId")
    local inProcessing = redis.call("ZSCORE", ns .. ":processing", jobId)
    local inDelayed = redis.call("ZSCORE", ns .. ":delayed", jobId)
    local inGroup = nil
    if gid then
      inGroup = redis.call("ZSCORE", ns .. ":g:" .. gid, jobId)
    end
    if (not inProcessing) and (not inDelayed) and (not inGroup) then
      if keepCompleted == 0 then
        redis.call("DEL", jobKey)
        redis.call("DEL", uniqueKey)
        redis.call("SET", uniqueKey, jobId)
      else
        -- Job hash exists and we're keeping completed jobs, ensure unique key exists
        redis.call("SET", uniqueKey, jobId)
        return jobId
      end
    else
      if keepCompleted == 0 then
        local status = redis.call("HGET", jobKey, "status")
        if status == "completed" then
          redis.call("DEL", jobKey)
          redis.call("DEL", uniqueKey)
          redis.call("SET", uniqueKey, jobId)
        else
          -- Job is still active, ensure unique key exists
          redis.call("SET", uniqueKey, jobId)
          return jobId
        end
      end
      local activeAgain = redis.call("ZSCORE", ns .. ":processing", jobId)
      local delayedAgain = redis.call("ZSCORE", ns .. ":delayed", jobId)
      local inGroupAgain = nil
      if gid then
        inGroupAgain = redis.call("ZSCORE", ns .. ":g:" .. gid, jobId)
      end
      local jobStillExists = redis.call("EXISTS", jobKey)
      if jobStillExists == 1 and (activeAgain or delayedAgain or inGroupAgain) then
        return jobId
      end
    end
  end
end

local gZ = ns .. ":g:" .. groupId

if not orderMs then
  orderMs = tonumber(redis.call("TIME")[1]) * 1000
end
local baseEpoch = 1704067200000
local relativeMs = orderMs - baseEpoch

-- Use date-based sequence key to auto-reset daily (prevents max int overflow)
local daysSinceEpoch = math.floor(orderMs / 86400000)
local seqKey = ns .. ":seq:" .. daysSinceEpoch
local seq = redis.call("INCR", seqKey)
local score = relativeMs * 1000 + seq

-- Get Redis server time for buffering logic (to be consistent with server time)
local timeResult = redis.call("TIME")
local now = tonumber(timeResult[1]) * 1000 + math.floor(tonumber(timeResult[2]) / 1000)

-- Use client timestamp for the job hash so timing calculations are accurate from client perspective
local timestamp = clientTimestamp or now

redis.call("HMSET", jobKey,
  "id", jobId,
  "groupId", groupId,
  "data", data,
  "attempts", "0",
  "maxAttempts", tostring(maxAttempts),
  "seq", tostring(seq),
  "timestamp", tostring(timestamp),
  "orderMs", tostring(orderMs),
  "score", tostring(score),
  "delayUntil", tostring(delayUntil)
)

-- Track group membership (idempotent)
redis.call("SADD", groupsKey, groupId)

-- Add job to group's sorted set first
redis.call("ZADD", gZ, score, jobId)

-- Determine job status for return value
local jobStatus = "waiting"

if delayUntil > 0 and delayUntil > now then
  -- Job is delayed, add to delayed set
  redis.call("ZADD", delayedKey, delayUntil, jobId)
  jobStatus = "delayed"
else
  -- Job is not delayed, determine if group should be ready or buffering
  -- Only use buffering if orderingDelayMs is meaningful (>= 100ms)
  -- Below 100ms, the overhead isn't worth it and sorted set ordering is sufficient
  if orderingDelayMs >= 100 then
    -- Check if group is currently locked (actively processing)
    local lockKey = ns .. ":lock:" .. groupId
    local isLocked = redis.call("EXISTS", lockKey)
    
    if isLocked == 1 then
      -- Group is actively processing, don't buffer - the group will be re-added to ready
      -- when the current job completes. The sorted set already maintains order.
      -- Don't add to ready queue, let complete script handle it
    else
      -- Group is not processing, check if we should buffer
      local groupBufferKey = ns .. ":buffer:" .. groupId
      local bufferStartTime = redis.call("GET", groupBufferKey)
      
      if not bufferStartTime then
        -- This is the first job arriving for an idle group
        -- Start buffering window
        redis.call("SET", groupBufferKey, tostring(now), "PX", orderingDelayMs + 1000)
        
        -- Add group to buffering set with score = when it should become ready
        local readyAt = now + orderingDelayMs
        redis.call("ZADD", ns .. ":buffering", readyAt, groupId)
      end
      -- If bufferStartTime exists, the group is already buffering - just accumulate jobs
    end
  else
    -- No buffering or orderingDelayMs too small - rely on sorted set ordering
    -- Make group ready immediately if this is the head job
    local head = redis.call("ZRANGE", gZ, 0, 0, "WITHSCORES")
    if head and #head >= 2 then
      local headScore = tonumber(head[2])
      redis.call("ZADD", readyKey, headScore, groupId)
    end
  end
end

-- Return job data to avoid race condition where job might be processed & cleaned up
-- before getJob() is called
return {jobId, groupId, data, "0", tostring(maxAttempts), tostring(timestamp), tostring(orderMs), tostring(delayUntil), jobStatus}


