feat: ZMS-6: Add redis cluster support

.github/workflows/test.yml

@@ -58,3 +58,74 @@ jobs:
               env:
                   APPCONF_dbs_redis: redis://127.0.0.1:6379/1
                   APPCONF_dbs_mongodb: 'mongodb://127.0.0.1:27017/wildduck-test?authSource=admin'
+
+    test-redis-cluster:
+        runs-on: ubuntu-latest
+        services:
+            redis-cluster:
+                image: grokzen/redis-cluster:7.0.10
+                env:
+                    IP: 127.0.0.1
+                    INITIAL_PORT: 7000
+                    MASTERS: 3
+                    SLAVES_PER_MASTER: 0
+                options: >-
+                    --health-cmd "redis-cli -c -p 7000 ping"
+                    --health-interval 10s
+                    --health-timeout 5s
+                    --health-retries 20
+                ports:
+                    - 7000:7000
+                    - 7001:7001
+                    - 7002:7002
+
+        steps:
+            - uses: actions/checkout@v6
+
+            - name: Start MongoDB
+              uses: supercharge/mongodb-github-action@1.12.1
+              with:
+                  mongodb-version: 6.0
+                  mongodb-port: 27017
+                  mongodb-db: wildduck-test
+
+            - name: Use Node.js 24.x
+              uses: actions/setup-node@v6
+              with:
+                  node-version: 24.x
+
+            - name: Install mongosh, mpop and redis-cli
+              run: |
+                  sudo apt-get update
+                  sudo apt-get install -y wget gnupg mpop redis-tools
+                  wget -qO - https://www.mongodb.org/static/pgp/server-6.0.asc | sudo apt-key add -
+                  echo "deb [ arch=amd64,arm64 ] https://repo.mongodb.org/apt/ubuntu jammy/mongodb-org/6.0 multiverse" | sudo tee /etc/apt/sources.list.d/mongodb-org-6.0.list
+                  sudo apt-get update
+                  sudo apt-get install -y mongodb-mongosh
+
+            - run: which mongosh
+            - run: which mpop
+            - run: which redis-cli
+            - run: npm install
+
+            - name: Wait for Redis Cluster
+              run: |
+                  for port in 7000 7001 7002; do
+                    for i in $(seq 1 60); do
+                      redis-cli -c -h 127.0.0.1 -p "$port" ping && break
+                      sleep 1
+                    done
+                  done
+                  redis-cli -c -h 127.0.0.1 -p 7000 cluster info | grep 'cluster_state:ok'
+
+            - name: Reset DB and Redis cluster
+              run: |
+                  mongosh --eval 'db.dropDatabase()' wildduck-test
+                  for port in 7000 7001 7002; do
+                    redis-cli -c -h 127.0.0.1 -p "$port" flushdb
+                  done
+
+            - name: Run tests against Redis cluster
+              run: NODE_ENV=test-cluster npx grunt
+              env:
+                  APPCONF_dbs_mongodb: 'mongodb://127.0.0.1:27017/wildduck-test?authSource=admin'

Gruntfile.js

@@ -1,7 +1,5 @@
 'use strict';
 
-process.env.NODE_ENV = 'test';
-
 module.exports = function (grunt) {
     // Project configuration.
     grunt.initConfig({

api.js

@@ -593,7 +593,7 @@ module.exports = done => {
     settingsRoutes(db, server, settingsHandler);
     healthRoutes(db, server, loggelf);
 
-    if (process.env.NODE_ENV === 'test') {
+    if (process.env.NODE_ENV === 'test' || process.env.NODE_ENV === 'test-cluster') {
         server.get(
             { name: 'api-methods', path: '/api-methods' },
             tools.responseWrapper(async (req, res) => {

bin/access-tokens

@@ -40,10 +40,7 @@ let argv = yargs
             await dbconnect();
 
             let accessToken = crypto.randomBytes(20).toString('hex');
-            let tokenHash = crypto
-                .createHash('sha256')
-                .update(accessToken)
-                .digest('hex');
+            let tokenHash = crypto.createHash('sha256').update(accessToken).digest('hex');
             let key = 'tn:token:' + tokenHash;
 
             let tokenData = {
@@ -63,11 +60,15 @@ let argv = yargs
                     .digest('hex')
             };
 
-            await db.redis
-                .multi()
-                .hmset(key, tokenData)
-                .sadd('tn:user:' + argv.user, tokenHash)
-                .exec();
+            if (db.redis.isCluster) {
+                await Promise.all([db.redis.hmset(key, tokenData), db.redis.sadd('tn:user:' + argv.user, tokenHash)]);
+            } else {
+                await db.redis
+                    .multi()
+                    .hmset(key, tokenData)
+                    .sadd('tn:user:' + argv.user, tokenHash)
+                    .exec();
+            }
 
             console.error('Generated access token for %s[%s]:', tokenData.user, tokenData.role);
 
@@ -92,10 +93,7 @@ let argv = yargs
 
             let accessToken = argv.token;
 
-            let tokenHash = crypto
-                .createHash('sha256')
-                .update(accessToken)
-                .digest('hex');
+            let tokenHash = crypto.createHash('sha256').update(accessToken).digest('hex');
 
             let key = 'tn:token:' + tokenHash;
 
@@ -132,13 +130,20 @@ let argv = yargs
                 process.exit();
             }
 
-            let query = await db.redis.multi().del('tn:user:' + user);
+            if (db.redis.isCluster) {
+                await db.redis.del('tn:user:' + user);
 
-            tokens.forEach(tokenHash => {
-                query = query.del('tn:token:' + tokenHash);
-            });
+                for (let tokenHash of tokens) {
+                    await db.redis.del('tn:token:' + tokenHash);
+                }
+            } else {
+                let query = db.redis.multi().del('tn:user:' + user);
+                tokens.forEach(tokenHash => {
+                    query = query.del('tn:token:' + tokenHash);
+                });
 
-            await query.exec();
+                await query.exec();
+            }
 
             console.error('Deleted %s tokens for %s', tokens.length, user);
 

config/dbs.toml

@@ -49,3 +49,25 @@ db = 3
 # [[redis.sentinels]]
 #   host="54.36.85.115"
 #   port=26379
+
+## Connect to Redis Cluster instead of a single master
+# [redis]
+# cluster = true
+# password = ""               # global/fallback password
+# [[redis.nodes]]
+#   host="54.36.85.113"
+#   port=26379
+#   password="" # per node password
+# [[redis.nodes]]
+#   host = "127.0.0.1"
+#   port = 7000
+# [[redis.nodes]]
+#   host = "127.0.0.1"
+#   port = 7001
+# [[redis.nodes]]
+#   host = "127.0.0.1"
+#   port = 7002
+# [[redis.nodes]]
+#   host="54.36.85.115"
+#   port=26379
+#   password=""

config/test-cluster.toml

@@ -0,0 +1,41 @@
+[api]
+port = 8080
+host = "127.0.0.1"
+secure = false
+
+[log]
+level = "silly"
+
+[dbs]
+# mongodb connection string for the main database
+mongo = "mongodb://127.0.0.1:27017/wildduck-test"
+
+dbname = "wildduck-test"
+
+[dbs.redis]
+cluster = true
+password = ""
+
+[[dbs.redis.nodes]]
+host = "127.0.0.1"
+port = 7000
+
+[[dbs.redis.nodes]]
+host = "127.0.0.1"
+port = 7001
+
+[[dbs.redis.nodes]]
+host = "127.0.0.1"
+port = 7002
+
+[imap]
+port = 9993
+host = "127.0.0.1"
+
+[lmtp]
+enabled = true
+port = 2424
+
+[pwned]
+enabled = true
+type = "softfail"

indexer.js

@@ -12,6 +12,7 @@ const counters = require('./lib/counters');
 const { ObjectId } = require('mongodb');
 const libmime = require('libmime');
 const punycode = require('punycode.js');
+const tools = require('./lib/tools');
 const { getClient } = require('./lib/elasticsearch');
 const { normalizeLoggelfMessage } = require('./lib/loggelf-message');
 
@@ -284,14 +285,15 @@ function indexingJob(esclient) {
 
             const dateKeyTdy = new Date().toISOString().substring(0, 10).replace(/-/g, '');
             const dateKeyYdy = new Date(Date.now() - 24 * 3600 * 1000).toISOString().substring(0, 10).replace(/-/g, '');
-            const tombstoneTdy = `indexer:tomb:${dateKeyTdy}`;
-            const tombstoneYdy = `indexer:tomb:${dateKeyYdy}`;
+            const tombstoneTag = tools.redisHashTag(db.redis, 'indexer:tomb');
+            const tombstoneTdy = `${tombstoneTag}:${dateKeyTdy}`;
+            const tombstoneYdy = `${tombstoneTag}:${dateKeyYdy}`;
 
             switch (data.action) {
                 case 'new': {
                     // check tombstone for race conditions (might be already deleted)
 
-                    let [[err1, isDeleted1], [err2, isDeleted2]] = await db.redis
+                    const [[err1, isDeleted1], [err2, isDeleted2]] = await db.redis
                         .multi()
                         .sismember(tombstoneTdy, data.message)
                         .sismember(tombstoneYdy, data.message)

lib/api/addresses.js

@@ -2397,11 +2397,12 @@ module.exports = (db, server, userHandler, settingsHandler) => {
 
             let response;
             try {
+                const addressKey = tools.redisHashTag(db.redis, addressData._id.toString());
                 response = await db.redis
                     .multi()
                     // sending counters are stored in Redis
-                    .get('wdf:' + addressData._id.toString())
-                    .ttl('wdf:' + addressData._id.toString())
+                    .get(`wdf:${addressKey}`)
+                    .ttl(`wdf:${addressKey}`)
                     .exec();
             } catch (err) {
                 // ignore
@@ -2565,11 +2566,12 @@ module.exports = (db, server, userHandler, settingsHandler) => {
 
             let response;
             try {
+                const addressKey = tools.redisHashTag(db.redis, addressData._id.toString());
                 response = await db.redis
                     .multi()
                     // sending counters are stored in Redis
-                    .get('wdf:' + addressData._id.toString())
-                    .ttl('wdf:' + addressData._id.toString())
+                    .get(`wdf:${addressKey}`)
+                    .ttl(`wdf:${addressKey}`)
                     .exec();
             } catch (err) {
                 // ignore

lib/api/mailboxes.js

@@ -257,7 +257,7 @@ module.exports = (db, server, mailboxHandler) => {
                 counterOps.push(
                     (async () => {
                         try {
-                            total = await getMailboxCounter(db, mailboxData._id);
+                            total = await getMailboxCounter(db, mailboxData._id, false, mailboxData.user);
                         } catch (err) {
                             // ignore
                         }
@@ -268,7 +268,7 @@ module.exports = (db, server, mailboxHandler) => {
                 counterOps.push(
                     (async () => {
                         try {
-                            unseen = await getMailboxCounter(db, mailboxData._id, 'unseen');
+                            unseen = await getMailboxCounter(db, mailboxData._id, 'unseen', mailboxData.user);
                         } catch (err) {
                             // ignore
                         }
@@ -533,13 +533,13 @@ module.exports = (db, server, mailboxHandler) => {
             let total, unseen;
 
             try {
-                total = await getMailboxCounter(db, mailboxData._id);
+                total = await getMailboxCounter(db, mailboxData._id, false, mailboxData.user);
             } catch (err) {
                 // ignore
             }
 
             try {
-                unseen = await getMailboxCounter(db, mailboxData._id, 'unseen');
+                unseen = await getMailboxCounter(db, mailboxData._id, 'unseen', mailboxData.user);
             } catch (err) {
                 // ignore
             }