JoobQ::Delivery::Scripts

ADMIT = Script.new(COMMON + ADMIT_BODY)

ADMIT_BODY = " local old = redis.call('HGET', KEYS[1], ARGV[1])\n if old then return {1, old} end\n if redis.call('HLEN', KEYS[1]) >= tonumber(ARGV[4]) then return {-1} end\n local e = envelope(ARGV[2], ARGV[1])\n if not e then return {-2} end\n e.due_at = clock() + tonumber(ARGV[3])\n if not integer(e.due_at) then return {-2} end\n write(e)\n return {3, e.due_at}"

ADMIT_MANY = Script.new((((COMMON + "\nlocal function one(ARGV)\n") + ADMIT_BODY) + "\nend\n") + " local results = {}\n for i = 3, #ARGV, 2 do\n results[#results + 1] = one({ARGV[i], ARGV[i+1], ARGV[1], ARGV[2]})\n end\n return results")

CLAIM = Script.new(COMMON + " if redis.call('EXISTS', KEYS[4]) == 1 then return {0} end\n local now = clock()\n local ids = redis.call('ZRANGEBYSCORE', KEYS[2], '-inf', now, 'LIMIT', 0, 100)\n for _, id in ipairs(ids) do\n local raw = redis.call('HGET', KEYS[1], id)\n if raw then\n local e = envelope(raw, id)\n if not e then return {-2} end\n if (e.state == 'ready' or e.state == 'scheduled') and e.due_at <= now then\n e.state = 'leased'\n e.attempt = e.attempt + 1\n e.owner = ARGV[1]\n e.token = ARGV[2]\n e.lease_until = now + tonumber(ARGV[3])\n if not integer(e.lease_until) then return {-2} end\n write(e)\n return {1, id, raw, e.attempt}\n end\n -- An orphan index must not hide healthy work behind the first page.\n redis.call('ZREM', KEYS[2], id)\n else\n redis.call('ZREM', KEYS[2], id)\n end\n end\n return {0}")

CLAIM_MANY = Script.new(COMMON + " if redis.call('EXISTS', KEYS[4]) == 1 then return {} end\n local now = clock()\n local results = {}\n local bytes = 0\n local ids = redis.call('ZRANGEBYSCORE', KEYS[2], '-inf', now, 'LIMIT', 0, tonumber(ARGV[4]))\n for _, id in ipairs(ids) do\n local raw = redis.call('HGET', KEYS[1], id)\n if raw then\n local e = envelope(raw, id)\n if not e or not integer(e.attempt + 1) or not integer(now + tonumber(ARGV[3])) then\n results[#results + 1] = {-2}\n elseif (e.state == 'ready' or e.state == 'scheduled') and e.due_at <= now then\n if bytes + #e.payload > tonumber(ARGV[5]) then break end\n bytes = bytes + #e.payload\n e.state = 'leased'\n e.attempt = e.attempt + 1\n e.owner = ARGV[1]\n e.token = ARGV[2] .. ':' .. tostring(#results + 1)\n e.lease_until = now + tonumber(ARGV[3])\n write(e)\n results[#results + 1] = {1, id, raw, e.attempt, e.token}\n else\n redis.call('ZREM', KEYS[2], id)\n end\n else\n redis.call('ZREM', KEYS[2], id)\n end\n end\n return results")

COMMON = " cjson.encode_number_precision(14)\n local function clock()\n local t = redis.call('TIME')\n return tonumber(t[1]) * 1000 + math.floor(tonumber(t[2]) / 1000)\n end\n local fields = {schema=true,id=true,payload=true,digest=true,state=true,attempt=true,\n token=true,owner=true,due_at=true,lease_until=true,traceparent=true}\n local states = {ready=true,leased=true,scheduled=true,completed=true,dead=true}\n local function integer(n)\n return type(n) == 'number' and n >= 0 and n < 99999999999999 and n == math.floor(n)\n end\n local function envelope(raw, id)\n local ok, e = pcall(cjson.decode, raw)\n if not ok or type(e) ~= 'table' or e.schema ~= 1 or e.id ~= id\n or type(e.payload) ~= 'string' or type(e.digest) ~= 'string'\n or #e.digest ~= 64 or not string.match(e.digest, '^[0-9a-f]+$')\n or not states[e.state] or not integer(e.attempt)\n or not integer(e.due_at) or not integer(e.lease_until) then return nil end\n for k, _ in pairs(e) do if not fields[k] then return nil end end\n if e.traceparent ~= nil and e.traceparent ~= cjson.null then\n if type(e.traceparent) ~= 'string' or #e.traceparent ~= 55\n or not string.match(e.traceparent, '^00%-[0-9a-f]+%-[0-9a-f]+%-[0-9a-f]+$')\n or string.sub(e.traceparent, 36, 36) ~= '-'\n or string.sub(e.traceparent, 53, 53) ~= '-'\n or string.sub(e.traceparent, 4, 35) == string.rep('0',32)\n or string.sub(e.traceparent, 37, 52) == string.rep('0',16) then return nil end\n end\n if (e.token ~= nil and e.token ~= cjson.null and type(e.token) ~= 'string')\n or (e.owner ~= nil and e.owner ~= cjson.null and type(e.owner) ~= 'string') then return nil end\n if e.state == 'leased' and (type(e.token) ~= 'string' or type(e.owner) ~= 'string' or e.lease_until <= 0) then return nil end\n return e\n end\n local function write(e)\n local raw = cjson.encode(e)\n -- Canonical intent MUST precede derived indexes; Lua errors do not roll back.\n redis.call('HSET', KEYS[1], e.id, raw)\n if e.state == 'ready' or e.state == 'scheduled' then\n redis.call('ZADD', KEYS[2], e.due_at, e.id)\n else\n redis.call('ZREM', KEYS[2], e.id)\n end\n if e.state == 'leased' then\n redis.call('ZADD', KEYS[3], e.lease_until, e.id)\n else\n redis.call('ZREM', KEYS[3], e.id)\n end\n return raw\n end"

QUARANTINE_CLAIM = Script.new(COMMON + " local raw = redis.call('HGET', KEYS[1], ARGV[1])\n if not raw then return 0 end\n local ok, e = pcall(cjson.decode, raw)\n if not ok or e.state ~= 'leased' or e.token ~= ARGV[2] or e.owner ~= ARGV[3]\n or e.attempt ~= tonumber(ARGV[4]) then return 0 end\n -- The client detected a checksum violation before executing the handler.\n -- Restore the original forensic bytes only while this claim still owns it.\n redis.call('HSET', KEYS[1], ARGV[1], ARGV[5])\n redis.call('HSET', KEYS[5], ARGV[1], ARGV[5])\n redis.call('ZREM', KEYS[2], ARGV[1])\n redis.call('ZREM', KEYS[3], ARGV[1])\n return 1")

SETTLE_MANY = Script.new((((COMMON + "\nlocal function one(ARGV)\n") + TRANSITION_BODY) + "\nend\n") + " local results = {}\n for i = 1, #ARGV, 8 do\n results[#results + 1] = one({ARGV[i], ARGV[i+1], ARGV[i+2], ARGV[i+3],\n ARGV[i+4], ARGV[i+5], ARGV[i+6], ARGV[i+7]})[1]\n end\n return results")

TRANSITION = Script.new(COMMON + TRANSITION_BODY)

TRANSITION_BODY = " local raw = redis.call('HGET', KEYS[1], ARGV[1])\n if not raw then return {0} end\n local e = envelope(raw, ARGV[1])\n if not e then return {-2} end\n -- Payload and digest are compared with the validated claim, so corruption\n -- after claiming cannot silently receive a successful acknowledgment.\n if e.payload ~= ARGV[7] or e.digest ~= ARGV[8] then return {-2} end\n local own = e.token == ARGV[2] and e.owner == ARGV[3] and e.attempt == tonumber(ARGV[4])\n if own and e.state == ARGV[5] and ARGV[5] ~= 'leased' then return {2} end\n local now = clock()\n if not own or e.state ~= 'leased' or e.lease_until <= now then return {0} end\n if ARGV[5] == 'leased' then\n if redis.call('EXISTS', KEYS[4]) == 1 then return {0} end\n e.lease_until = now + tonumber(ARGV[6])\n else\n e.state = ARGV[5]\n e.lease_until = 0\n e.due_at = now + tonumber(ARGV[6])\n end\n if not integer(e.lease_until) or not integer(e.due_at) then return {-2} end\n write(e)\n return {1}"