refactor(commnads): reuse getOrSetMaxEvents when missing

src/commands/addDelayedJob-6.lua

@@ -55,12 +55,11 @@ local parent = args[8]
 local parentData
 
 -- Includes
---- @include "includes/storeJob"
 --- @include "includes/addDelayMarkerIfNeeded"
+--- @include "includes/getOrSetMaxEvents"
 --- @include "includes/isQueuePaused"
---- @include "includes/getNextDelayedTimestamp"
+--- @include "includes/storeJob"
 --- @include "includes/updateExistingJobsParent"
---- @include "includes/getOrSetMaxEvents"
 
 if parentKey ~= nil then
     if rcall("EXISTS", parentKey) ~= 1 then return -5 end

src/commands/addStandardJob-7.lua

@@ -60,8 +60,8 @@ local parentData
 -- Includes
 --- @include "includes/storeJob"
 --- @include "includes/updateExistingJobsParent"
---- @include "includes/getTargetQueueList"
 --- @include "includes/getOrSetMaxEvents"
+--- @include "includes/getTargetQueueList"
 
 if parentKey ~= nil then
     if rcall("EXISTS", parentKey) ~= 1 then return -5 end

src/commands/includes/getJobsInZset.lua

@@ -1,7 +1,9 @@
--- We use ZRANGEBYSCORE to make the case where we're deleting a limited number
--- of items in a sorted set only run a single iteration. If we simply used
--- ZRANGE, we may take a long time traversing through jobs that are within the
--- grace period.
+--[[
+  We use ZRANGEBYSCORE to make the case where we're deleting a limited number
+  of items in a sorted set only run a single iteration. If we simply used
+  ZRANGE, we may take a long time traversing through jobs that are within the
+  grace period.
+]]
 local function getJobsInZset(zsetKey, rangeEnd, limit)
   if limit > 0 then
     return rcall("ZRANGEBYSCORE", zsetKey, 0, rangeEnd, "LIMIT", 0, limit)

src/commands/includes/getNextDelayedTimestamp.lua

@@ -1,6 +1,6 @@
 --[[
   Function to return the next delayed job timestamp.
-]] 
+]]
 local function getNextDelayedTimestamp(delayedKey)
   local result = rcall("ZRANGE", delayedKey, 0, 0, "WITHSCORES")
   if #result then

src/commands/includes/getOrSetMaxEvents.lua

@@ -1,4 +1,6 @@
-
+--[[
+  Function to get max events value or set by default 10000.
+]]
 local function getOrSetMaxEvents(metaKey)
     local maxEvents = rcall("HGET", metaKey, "opts.maxLenEvents")
     if not maxEvents then

src/commands/includes/getRateLimitTTL.lua

@@ -1,3 +1,6 @@
+--[[
+  Function to get current rate limit ttl.
+]]
 local function getRateLimitTTL(maxJobs, rateLimiterKey)
   if maxJobs and maxJobs <= tonumber(rcall("GET", rateLimiterKey) or 0) then
     local pttl = rcall("PTTL", rateLimiterKey)

src/commands/includes/trimEvents.lua

@@ -2,8 +2,11 @@
   Function to trim events, default 10000.
 ]]
 
+-- Includes
+--- @include "getOrSetMaxEvents"
+
 local function trimEvents(metaKey, eventStreamKey)
-  local maxEvents = rcall("HGET", metaKey, "opts.maxLenEvents")
+  local maxEvents = getOrSetMaxEvents(metaKey)
   if maxEvents ~= false then
     rcall("XTRIM", eventStreamKey, "MAXLEN", "~", maxEvents)
   else

src/commands/includes/updateExistingJobsParent.lua

@@ -1,11 +1,13 @@
---- @include "updateParentDepsIfNeeded"
-
 --[[
     This function is used to update the parent's dependencies if the job
     is already completed and about to be ignored. The parent must get its
     dependencies updated to avoid the parent job being stuck forever in 
     the waiting-children state.
 ]]
+
+-- Includes
+--- @include "updateParentDepsIfNeeded"
+
 local function updateExistingJobsParent(parentKey, parent, parentData,
                                         parentDependenciesKey, completedKey,
                                         jobIdKey, jobId, timestamp)

src/commands/moveJobsToWait-6.lua

@@ -26,6 +26,7 @@ local rcall = redis.call;
 
 -- Includes
 --- @include "includes/batches"
+--- @include "includes/getOrSetMaxEvents"
 --- @include "includes/getTargetQueueList"
 
 local metaKey = KEYS[6]
@@ -46,7 +47,7 @@ if (#jobs > 0) then
         end
     end
 
-    local maxEvents = rcall("HGET", metaKey, "opts.maxLenEvents") or 10000
+    local maxEvents = getOrSetMaxEvents(metaKey)
 
     for i, key in ipairs(jobs) do
         -- Emit waiting event

src/commands/moveToDelayed-7.lua

@@ -30,6 +30,7 @@ local rcall = redis.call
 
 -- Includes
 --- @include "includes/addDelayMarkerIfNeeded"
+--- @include "includes/getOrSetMaxEvents"
 --- @include "includes/isQueuePaused"
 
 local jobKey = KEYS[5]
@@ -58,7 +59,7 @@ if rcall("EXISTS", jobKey) == 1 then
 
     rcall("HSET", jobKey, "delay", ARGV[6])
 
-    local maxEvents = rcall("HGET", metaKey, "opts.maxLenEvents") or 10000
+    local maxEvents = getOrSetMaxEvents(metaKey)
 
     rcall("ZADD", delayedKey, score, jobId)
     rcall("XADD", KEYS[6], "MAXLEN", "~", maxEvents, "*", "event", "delayed",

src/commands/removeJob-1.lua

@@ -16,6 +16,7 @@ local rcall = redis.call
 
 -- Includes
 --- @include "includes/destructureJobKey"
+--- @include "includes/getOrSetMaxEvents"
 --- @include "includes/isLocked"
 --- @include "includes/removeJobFromAnyState"
 --- @include "includes/removeParentDependencyKey"
@@ -54,7 +55,7 @@ local function removeJob( prefix, jobId, parentKey, removeChildren)
     local prev = removeJobFromAnyState(prefix, jobId)
 
     if rcall("DEL", jobKey, jobKey .. ":logs", jobKey .. ":dependencies", jobKey .. ":processed") > 0 then
-        local maxEvents = rcall("HGET", prefix .. "meta", "opts.maxLenEvents") or 10000
+        local maxEvents = getOrSetMaxEvents(prefix .. "meta")
         rcall("XADD", prefix .. "events", "MAXLEN", "~", maxEvents, "*", "event", "removed",
             "jobId", jobId, "prev", prev)
     end

src/commands/retryJob-10.lua

@@ -31,6 +31,7 @@ local rcall = redis.call
 
 -- Includes
 --- @include "includes/addJobWithPriority"
+--- @include "includes/getOrSetMaxEvents"
 --- @include "includes/getTargetQueueList"
 --- @include "includes/promoteDelayedJobs"
 
@@ -65,7 +66,7 @@ if rcall("EXISTS", KEYS[4]) == 1 then
 
   rcall("HINCRBY", KEYS[4], "atm", 1)
 
-  local maxEvents = rcall("HGET", KEYS[5], "opts.maxLenEvents") or 10000
+  local maxEvents = getOrSetMaxEvents(KEYS[5])
 
   -- Emit waiting event
   rcall("XADD", KEYS[6], "MAXLEN", "~", maxEvents, "*", "event", "waiting",

src/commands/updateProgress-3.lua

@@ -18,8 +18,11 @@
 ]]
 local rcall = redis.call
 
+-- Includes
+--- @include "includes/getOrSetMaxEvents"
+
 if rcall("EXISTS", KEYS[1]) == 1 then -- // Make sure job exists
-    local maxEvents = rcall("HGET", KEYS[3], "opts.maxLenEvents") or 10000
+    local maxEvents = getOrSetMaxEvents(KEYS[3])
 
     rcall("HSET", KEYS[1], "progress", ARGV[2])
     rcall("XADD", KEYS[2], "MAXLEN", "~", maxEvents, "*", "event", "progress",

tests/test_flow.ts

@@ -2469,16 +2469,10 @@ describe('flows', () => {
 
   it('should add meta key to both parents and children', async () => {
     const name = 'child-job';
-    const values = [
-      { idx: 0, bar: 'something' },
-      { idx: 1, baz: 'something' },
-      { idx: 2, qux: 'something' },
-    ];
-
     const topQueueName = `top-queue-${v4()}`;
 
     const flow = new FlowProducer({ connection, prefix });
-    const tree = await flow.add({
+    await flow.add({
       name: 'root-job',
       queueName: topQueueName,
       data: {},
@@ -2501,10 +2495,10 @@ describe('flows', () => {
 
     const client = await flow.client;
     const metaTop = await client.hgetall(`${prefix}:${topQueueName}:meta`);
-    expect(metaTop).to.have.be.deep.equal({ 'opts.maxLenEvents': '10000' });
+    expect(metaTop).to.deep.include({ 'opts.maxLenEvents': '10000' });
 
     const metaChildren = await client.hgetall(`${prefix}:${queueName}:meta`);
-    expect(metaChildren).to.have.be.deep.equal({
+    expect(metaChildren).to.deep.include({
       'opts.maxLenEvents': '10000',
     });