Atomic expires

packages/backend-core/src/redis/redis.ts

@@ -289,7 +289,24 @@ class RedisWrapper {
       acc[addDbPrefix(this._db, key)] = value
       return acc
     }, {} as Record<string, any>)
-    await client.mset(dataToStore)
+
+    const luaScript = `
+        for i, key in ipairs(KEYS) do
+            redis.call('MSET', key, ARGV[i])
+            ${
+              expirySeconds !== null
+                ? `redis.call('EXPIRE', key, ARGV[#ARGV])`
+                : ""
+            }
+        end
+        `
+    const keys = Object.keys(dataToStore)
+    let values = Object.values(dataToStore)
+    if (expirySeconds !== null) {
+      values.push(expirySeconds)
+    }
+
+    await client.eval(luaScript, keys.length, ...keys, ...values)
   }
 
   async getTTL(key: string) {

packages/backend-core/src/redis/tests/redis.spec.ts

@@ -37,5 +37,24 @@ describe("redis", () => {
 
       expect(await redis.keys("*")).toHaveLength(10)
     })
+
+    it("a bulk store can be persisted with TTL", async () => {
+      const ttl = 500
+      const data = generator
+        .unique(() => generator.word(), 10)
+        .reduce((acc, key) => {
+          acc[key] = generator.word()
+          return acc
+        }, {} as Record<string, string>)
+
+      await redis.bulkStore(data, ttl)
+
+      for (const [key, value] of Object.entries(data)) {
+        expect(await redis.get(key)).toEqual(value)
+        expect(await redis.getTTL(key)).toEqual(ttl)
+      }
+
+      expect(await redis.keys("*")).toHaveLength(10)
+    })
   })
 })