feat(sending): Support 'Idempotency-Key' header for /submit message requests and 'X-EE-Idempotency-Key' SMTP header to avoid sending duplicate emails

andris9 · andris9 · commit 1423135b7049 · 2025-04-10T14:41:09.000+03:00
diff --git a/lib/db.js b/lib/db.js
@@ -128,6 +128,7 @@ const hSetNewScript = fs.readFileSync(pathlib.join(__dirname, 'lua/h-set-new.lua
 const hIncrbyExistsScript = fs.readFileSync(pathlib.join(__dirname, 'lua/h-incrby-exists.lua'), 'utf-8');
 const eeListAddScript = fs.readFileSync(pathlib.join(__dirname, 'lua/ee-list-add.lua'), 'utf-8');
 const eeListRemoveScript = fs.readFileSync(pathlib.join(__dirname, 'lua/ee-list-remove.lua'), 'utf-8');
+const eeGetIdempotencyScript = fs.readFileSync(pathlib.join(__dirname, 'lua/ee-get-idempotency.lua'), 'utf-8');
 
 redis.defineCommand('zExpunge', {
     numberOfKeys: 2,
@@ -204,6 +205,11 @@ redis.defineCommand('eeListRemove', {
     lua: eeListRemoveScript
 });
 
+redis.defineCommand('eeGetIdempotency', {
+    numberOfKeys: 1,
+    lua: eeGetIdempotencyScript
+});
+
 module.exports.redis = redis;
 module.exports.notifyQueue = notifyQueue;
 module.exports.submitQueue = submitQueue;
diff --git a/lib/email-client/base-client.js b/lib/email-client/base-client.js
@@ -1,6 +1,6 @@
 'use strict';
 
-const { parentPort } = require('worker_threads');
+const { parentPort, threadId: workerThreadId } = require('worker_threads');
 const crypto = require('crypto');
 const logger = require('../logger');
 const { webhooks: Webhooks } = require('../webhooks');
@@ -43,7 +43,8 @@ const {
     readEnvValue,
     emitChangeEvent,
     filterEmptyObjectValues,
-    resolveCredentials
+    resolveCredentials,
+    getDateBuckets
 } = require('../tools');
 
 const {
@@ -84,6 +85,8 @@ async function metricsMeta(meta, logger, key, method, ...args) {
     }
 }
 
+const pendingIdempotencyOperations = new Map();
+
 class BaseClient {
     constructor(account, options) {
         this.account = account;
@@ -619,7 +622,136 @@ class BaseClient {
         return oauthCredentials;
     }
 
+    async checkIdempotencyKey(objName, idempotencyKey) {
+        if (!idempotencyKey) {
+            return null;
+        }
+
+        const idempotencyKeyName = objName ? `${objName}/${idempotencyKey}` : idempotencyKey;
+
+        // check last 24-48 hours, so probably will return 2 keys, at rare cases 1
+        const { bucketKeys } = getDateBuckets(1 * 24 * 3600);
+
+        let idempotencyResultStr = await this.redis.eeGetIdempotency(
+            `${REDIS_PREFIX}idempotency:bucket:`,
+            idempotencyKeyName,
+            this.runIndex,
+            workerThreadId,
+            bucketKeys.join(',')
+        );
+
+        let idempotencyData;
+        try {
+            idempotencyData = JSON.parse(idempotencyResultStr);
+            idempotencyData.idempotencyKey = idempotencyKey;
+            idempotencyData.idempotencyKeyName = idempotencyKeyName;
+        } catch (err) {
+            this.logger.error({ msg: 'Failed to parse idempotency data', idempotencyKey, cachedValue: idempotencyResultStr, err });
+        }
+
+        if (idempotencyData?.status === 'new') {
+            if (pendingIdempotencyOperations.has(idempotencyKeyName)) {
+                let error = new Error('Cancelling pending operation');
+                for (let promise of pendingIdempotencyOperations.get(idempotencyKeyName)) {
+                    promise.reject(error);
+                }
+            }
+            pendingIdempotencyOperations.set(idempotencyKeyName, []);
+        }
+
+        // use existing response
+        switch (idempotencyData.status) {
+            case 'completed':
+                idempotencyData.returnValue = Object.assign({}, idempotencyData.result, {
+                    idempotency: { key: idempotencyData.idempotencyKey, status: 'HIT' }
+                });
+                break;
+            case 'pending': {
+                let queueResult = await new Promise((resolve, reject) => {
+                    pendingIdempotencyOperations.get(idempotencyData.idempotencyKeyName).push({ resolve, reject });
+                });
+                idempotencyData.returnValue = Object.assign({}, queueResult, { idempotency: { key: idempotencyData.idempotencyKey, status: 'HIT' } });
+                break;
+            }
+        }
+
+        return idempotencyData || null;
+    }
+
+    async updateIdempotencyData(idempotencyData, result) {
+        if (idempotencyData?.bucketKey && idempotencyData?.idempotencyKeyName) {
+            // update status and result
+            try {
+                await this.redis.hset(
+                    idempotencyData?.bucketKey,
+                    idempotencyData.idempotencyKeyName,
+                    JSON.stringify({
+                        status: 'completed',
+                        runIndex: idempotencyData.runIndex,
+                        threadId: idempotencyData.threadId,
+                        result
+                    })
+                );
+            } catch (err) {
+                this.logger.error({
+                    msg: 'Failed to update idempotency data',
+                    idempotencyKey: idempotencyData.idempotencyKey,
+                    err
+                });
+            }
+
+            for (let promise of pendingIdempotencyOperations.get(idempotencyData.idempotencyKeyName)) {
+                promise.resolve(result);
+            }
+        }
+    }
+
+    async clearIdempotencyData(idempotencyData, error) {
+        if (idempotencyData?.status === 'new' && idempotencyData?.bucketKey && idempotencyData?.idempotencyKeyName) {
+            // delete failed attempt information
+            try {
+                await this.redis.hdel(idempotencyData?.bucketKey, idempotencyData.idempotencyKeyName);
+            } catch (err) {
+                this.logger.error({
+                    msg: 'Failed to clear idempotency data',
+                    idempotencyKey: idempotencyData.idempotencyKey,
+                    err
+                });
+            }
+
+            for (let promise of pendingIdempotencyOperations.get(idempotencyData.idempotencyKeyName)) {
+                promise.reject(error);
+            }
+        }
+    }
+
     async queueMessage(data, meta, connectionOptions) {
+        let idempotencyData;
+
+        if (meta.idempotencyKey) {
+            idempotencyData = await this.checkIdempotencyKey(`mq/${this.account}`, meta.idempotencyKey);
+            if (idempotencyData?.returnValue) {
+                return idempotencyData?.returnValue;
+            }
+        }
+
+        let queueResult;
+        try {
+            queueResult = await this.queueMessageHandler(data, meta, connectionOptions);
+            await this.updateIdempotencyData(idempotencyData, queueResult);
+        } catch (err) {
+            await this.clearIdempotencyData(idempotencyData, err);
+            throw err;
+        }
+
+        if (idempotencyData?.status === 'new') {
+            return Object.assign({}, queueResult, { idempotency: { key: idempotencyData.idempotencyKey, status: 'MISS' } });
+        }
+
+        return queueResult;
+    }
+
+    async queueMessageHandler(data, meta, connectionOptions) {
         let accountData = await this.accountObject.loadAccountData();
 
         let gatewayData;
diff --git a/lib/lua/ee-get-idempotency.lua b/lib/lua/ee-get-idempotency.lua
@@ -0,0 +1,52 @@
+
+local bucketKeyPrefix = KEYS[1];
+local idempotencyKey = ARGV[1];
+
+-- Use run index and thread ID check to determine if previously pending job has been terminated
+-- EmailEngine assigns a specific thread to process all actions for the same user
+local runIndex = tonumber(ARGV[2]);
+local threadId = tonumber(ARGV[3]);
+
+local buckets = ARGV[4];
+
+local lastBucket = nil;
+
+for bucket in string.gmatch(buckets, "([^,]+)") do
+    local bucketKey = bucketKeyPrefix .. bucket;
+    if redis.call("HEXISTS", bucketKey, idempotencyKey) == 1 then
+        local existingValue = redis.call("HGET", bucketKeyPrefix .. bucket, idempotencyKey);
+
+        local parsedValue = cjson.decode(existingValue);
+        local existingStatus = parsedValue["status"];
+        local existingRunIndex = parsedValue["runIndex"];
+        local existingThreadId = parsedValue["threadId"];
+
+        if existingStatus == "pending" and (existingRunIndex < runIndex or existingThreadId ~= threadId) then
+            -- found match but ignore it
+            redis.log( redis.LOG_NOTICE, "EE: Ignoring pending task with old run index: " .. existingValue .. " Current run index: " .. tostring(runIndex).. " Current thread ID: " .. tostring(threadId));
+        else
+            parsedValue['bucketKey'] = bucketKey;
+            return existingValue;
+        end
+
+    end
+
+    lastBucket = bucket;
+end
+
+-- No idempotency key found, create a new entry to the newest bucket
+local bucketKey = bucketKeyPrefix .. lastBucket;
+
+local newValue = {
+    ['status'] = 'pending',
+    ['runIndex'] = runIndex,
+    ['threadId'] = threadId
+};
+
+redis.call("HSET", bucketKey, idempotencyKey, cjson.encode(newValue));
+redis.call("EXPIRE", bucketKey, 24 * 3600);
+
+-- return with a "new" status
+newValue['status'] = 'new';
+newValue['bucketKey'] = bucketKey;
+return cjson.encode(newValue)
diff --git a/lib/tools.js b/lib/tools.js
@@ -531,13 +531,11 @@ module.exports = {
         return entry;
     },
 
-    async getCounterValues(redis, seconds) {
-        seconds = Number(seconds) || 3600;
-
+    getDateBuckets(seconds) {
         let now = new Date();
         let startTime = new Date(now.getTime() - seconds * 1000);
 
-        let hashKeys = [];
+        let bucketKeys = [];
 
         // find out all the date buckets we need to check for
         let endDateStr = `${now
@@ -550,18 +548,26 @@ module.exports = {
         let startTimeStr = `${startTime
             .toISOString()
             // bucket includes 1 minute
-            .substr(0, 16)
+            .substring(0, 16)
             .replace(/[^0-9]+/g, '')}`;
 
         while (dateStr < endDateStr) {
             dateStr = `${hashTime
                 .toISOString()
                 .substr(0, 10)
                 .replace(/[^0-9]+/g, '')}`;
-            hashKeys.push(dateStr);
+            bucketKeys.push(dateStr);
             hashTime = new Date(hashTime.getTime() + 24 * 3600 * 1000);
         }
 
+        return { bucketKeys, startTimeStr };
+    },
+
+    async getCounterValues(redis, seconds) {
+        seconds = Number(seconds) || 3600;
+
+        const { bucketKeys: hashKeys, startTimeStr } = module.exports.getDateBuckets(seconds);
+
         // list potential counter keys
         let statUpdateKeys = await redis.smembers(`${REDIS_PREFIX}stats:keys`);
 
diff --git a/workers/api.js b/workers/api.js
@@ -5272,7 +5272,10 @@ const init = async () => {
             });
 
             try {
-                return await accountObject.queueMessage(request.payload, { source: 'api' });
+                return await accountObject.queueMessage(request.payload, {
+                    source: 'api',
+                    idempotencyKey: request.headers['idempotency-key']
+                });
             } catch (err) {
                 request.logger.error({ msg: 'API request failed', err });
                 if (Boom.isBoom(err)) {
@@ -5318,6 +5321,24 @@ const init = async () => {
                     account: accountIdSchema.required()
                 }),
 
+                headers: Joi.object({
+                    'x-ee-timeout': Joi.number()
+                        .integer()
+                        .min(0)
+                        .max(2 * 3600 * 1000)
+                        .optional()
+                        .description(`Override the \`EENGINE_TIMEOUT\` environment variable for a single API request (in milliseconds)`)
+                        .label('X-EE-Timeout'),
+                    'idempotency-key': Joi.string()
+                        .min(0)
+                        .max(1024)
+                        .optional()
+                        .description(
+                            'A unique identifier provided by the client to ensure that repeated requests with the same key are processed only once. The value should be unique per operation and can be up to 1024 characters in length.'
+                        )
+                        .label('Idempotency-Key')
+                }).unknown(),
+
                 payload: Joi.object({
                     reference: messageReferenceSchema,
 
diff --git a/workers/smtp.js b/workers/smtp.js
@@ -222,6 +222,12 @@ function processMessage(stream, session, meta) {
         if (requestedAccount) {
             meta.requestedAccount = requestedAccount;
         }
+
+        let idempotencyKey = headers.getFirst('x-ee-idempotency-key');
+        headers.remove('x-ee-idempotency-key');
+        if (idempotencyKey) {
+            meta.idempotencyKey = idempotencyKey;
+        }
     });
 
     stream.once('error', err => joiner.emit('error', err));
@@ -372,7 +378,10 @@ async function init() {
                     };
 
                     accountObject
-                        .queueMessage(payload, { source: 'smtp' })
+                        .queueMessage(payload, {
+                            source: 'smtp',
+                            idempotencyKey: messageMeta.idempotencyKey
+                        })
                         .then(res => {
                             // queued for later
                             metrics(logger, 'events', 'inc', {
@@ -384,7 +393,8 @@ async function init() {
                                 account: session.user,
                                 messageId: res.messageId,
                                 sendAt: res.sendAt,
-                                queueId: res.queueId
+                                queueId: res.queueId,
+                                idempotency: res.idempotency
                             });
 
                             return callback(null, `Message queued for delivery as ${res.queueId} (${new Date(res.sendAt).toISOString()})`);
