apalis-redis 1.0.0

Background task processing for rust using apalis and redis
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94

-- KEYS[1]: the task data hash
-- KEYS[2]: the active task list
-- KEYS[3]: the signal list
-- KEYS[4]: the metadata prefix (e.g. "task_meta")
-- KEYS[5]: the scheduled set (for delayed tasks)
-- KEYS[6]: the idempotency key set (e.g. "task_idempotency")

-- ARGV[1]: the idempotency_ttl
-- ARGV[2]: emit pubsub events
-- ARGV[3..]: [task_id1, task_data1, attempts1, max_attempts1, metadata_json1, schedule_time1, idempotency_key1,
--        task_id2, task_data2, attempts2, max_attempts2, metadata_json2, schedule_time2, idempotency_key2, ...]
local newly_enqueued = 0
local newly_scheduled = 0

local idempotency_ttl = ARGV[1]
local emit_events = ARGV[2] == "1"

for i = 3, #ARGV, 7 do
    local task_id = ARGV[i]
    local task_data = ARGV[i + 1]
    local attempts = ARGV[i + 2]
    local max_attempts = ARGV[i + 3]
    local metadata_json = ARGV[i + 4]
    local schedule_time = tonumber(ARGV[i + 5])
    local idempotency_key = ARGV[i + 6] -- empty string "" means no idempotency key

    if task_id and task_data then

        -- If an idempotency key is provided, attempt to claim it.
        -- setnx returns 0 if already exists, meaning this task was already submitted.
        local allowed = true
        if idempotency_key and #idempotency_key > 0 then
            local idem_field = KEYS[6] .. ':' .. idempotency_key
            local claimed = redis.call("setnx", idem_field, task_id)
            if claimed == 0 then
                allowed = false
            else
                if idempotency_ttl and #idempotency_ttl > 0 then
                    redis.call("expire", idem_field, idempotency_ttl)
                end
            end
        end

        if allowed then
            local set = redis.call("hsetnx", KEYS[1], task_id, task_data)

            if set == 1 then
                local meta_key = KEYS[4] .. ':' .. task_id

                -- Base metadata
                local meta = {"attempts", tonumber(attempts), "max_attempts", tonumber(max_attempts), "status",
                              "Pending"}

                -- Store the idempotency key in metadata for traceability
                table.insert(meta, "idempotency_key")
                table.insert(meta, idempotency_key)

                -- Merge in any JSON metadata
                if metadata_json and #metadata_json > 0 then
                    local ok, decoded = pcall(cjson.decode, metadata_json)
                    if ok and type(decoded) == "table" then
                        for k, v in pairs(decoded) do
                            if type(v) == "table" then
                                v = cjson.encode(v)
                            end
                            table.insert(meta, k)
                            table.insert(meta, tostring(v))
                        end
                    end
                end

                redis.call("hmset", meta_key, unpack(meta))

                if schedule_time and schedule_time > 0 then
                    redis.call("zadd", KEYS[5], schedule_time, task_id)
                    newly_scheduled = newly_scheduled + 1
                else
                    redis.call("rpush", KEYS[2], task_id)
                    if emit_events then
                        redis.call("publish", "tasks:" .. KEYS[2] .. ':available', task_id)
                    end
                    newly_enqueued = newly_enqueued + 1
                end
            end
        end
    end
end

if newly_enqueued > 0 then
    redis.call("del", KEYS[3])
    redis.call("lpush", KEYS[3], 1)
end

return {newly_enqueued, newly_scheduled}