fix: Refactor Redis keys to prevent multiple messages on concurrent requests

apps/api/src/jobs/domain-verification.ts

@@ -122,9 +122,13 @@ export async function checkDomainVerifications() {
 
           // Send email notification about domain verified
           try {
+            // Use SETNX to atomically check and set the flag (prevents race conditions)
             const cacheKey = Keys.Domain.verifiedEmail(dbDomain.id);
-            const alreadySent = await redis.get(cacheKey);
+            const ttl = 604800; // 7 days
-            if (alreadySent !== '1') {
+
+            // SETNX returns 1 if key was set (didn't exist), 0 if key already existed
+            const wasSet = await redis.set(cacheKey, '1', 'EX', ttl, 'NX');
+            if (wasSet) {
               const members = await MembershipService.getMembers(dbDomain.projectId);
               const emails = members.map(m => m.email);
               if (emails.length > 0) {
@@ -138,7 +142,6 @@ export async function checkDomainVerifications() {
                 await Promise.all(
                   emails.map(email => sendPlatformEmail(email, 'Domain Verified Successfully', template)),
                 );
-                await redis.setex(cacheKey, 604800, '1'); // 7 days
               }
             }
           } catch (error) {
@@ -156,12 +159,17 @@ export async function checkDomainVerifications() {
 
           // Send email notification about domain verification failed
           try {
+            // Use SETNX to atomically check and set the flag (prevents race conditions)
             const now = new Date();
             const year = now.getFullYear();
             const month = String(now.getMonth() + 1).padStart(2, '0');
             const cacheKey = Keys.Domain.unverifiedEmail(dbDomain.id, year, month);
-            const alreadySent = await redis.get(cacheKey);
+            const endOfMonth = new Date(now.getFullYear(), now.getMonth() + 1, 1);
-            if (alreadySent !== '1') {
+            const ttl = Math.floor((endOfMonth.getTime() - now.getTime()) / 1000);
+
+            // SETNX returns 1 if key was set (didn't exist), 0 if key already existed
+            const wasSet = await redis.set(cacheKey, '1', 'EX', ttl, 'NX');
+            if (wasSet) {
               const members = await MembershipService.getMembers(dbDomain.projectId);
               const emails = members.map(m => m.email);
               if (emails.length > 0) {
@@ -175,9 +183,6 @@ export async function checkDomainVerifications() {
                 await Promise.all(
                   emails.map(email => sendPlatformEmail(email, 'Domain Verification Failed', template)),
                 );
-                const endOfMonth = new Date(now.getFullYear(), now.getMonth() + 1, 1);
-                const ttl = Math.floor((endOfMonth.getTime() - now.getTime()) / 1000);
-                await redis.setex(cacheKey, ttl, '1');
               }
             }
           } catch (error) {

apps/api/src/services/BillingLimitService.ts

@@ -580,8 +580,15 @@ export class BillingLimitService {
       const month = String(now.getMonth() + 1).padStart(2, '0');
       const cacheKey = Keys.Billing.warningEmail(projectId, sourceType, year, month);
 
-      const alreadySent = await redis.get(cacheKey);
+      // Use SETNX (SET if Not eXists) to atomically check and set the flag
-      if (alreadySent === '1') {
+      // This prevents race conditions where multiple concurrent requests could all pass the check
+      const endOfMonth = new Date(now.getFullYear(), now.getMonth() + 1, 1);
+      const ttl = Math.floor((endOfMonth.getTime() - now.getTime()) / 1000);
+
+      // SETNX returns 1 if key was set (didn't exist), 0 if key already existed
+      const wasSet = await redis.set(cacheKey, '1', 'EX', ttl, 'NX');
+      if (!wasSet) {
+        // Email was already sent this month
         return;
       }
 
@@ -603,11 +610,6 @@ export class BillingLimitService {
       });
 
       await Promise.all(emails.map(email => sendPlatformEmail(email, 'Billing Limit Warning', template)));
-
-      // Mark that we've sent the warning email (expires at end of month)
-      const endOfMonth = new Date(now.getFullYear(), now.getMonth() + 1, 1);
-      const ttl = Math.floor((endOfMonth.getTime() - now.getTime()) / 1000);
-      await redis.setex(cacheKey, ttl, '1');
     } catch (error) {
       signale.error(`[BILLING_LIMIT] Failed to send warning email:`, error);
     }
@@ -630,8 +632,15 @@ export class BillingLimitService {
       const month = String(now.getMonth() + 1).padStart(2, '0');
       const cacheKey = Keys.Billing.limitEmail(projectId, sourceType, year, month);
 
-      const alreadySent = await redis.get(cacheKey);
+      // Use SETNX (SET if Not eXists) to atomically check and set the flag
-      if (alreadySent === '1') {
+      // This prevents race conditions where multiple concurrent requests could all pass the check
+      const endOfMonth = new Date(now.getFullYear(), now.getMonth() + 1, 1);
+      const ttl = Math.floor((endOfMonth.getTime() - now.getTime()) / 1000);
+
+      // SETNX returns 1 if key was set (didn't exist), 0 if key already existed
+      const wasSet = await redis.set(cacheKey, '1', 'EX', ttl, 'NX');
+      if (!wasSet) {
+        // Email was already sent this month
         return;
       }
 
@@ -652,11 +661,6 @@ export class BillingLimitService {
       });
 
       await Promise.all(emails.map(email => sendPlatformEmail(email, 'Billing Limit Exceeded', template)));
-
-      // Mark that we've sent the limit email (expires at end of month)
-      const endOfMonth = new Date(now.getFullYear(), now.getMonth() + 1, 1);
-      const ttl = Math.floor((endOfMonth.getTime() - now.getTime()) / 1000);
-      await redis.setex(cacheKey, ttl, '1');
     } catch (error) {
       signale.error(`[BILLING_LIMIT] Failed to send limit exceeded email:`, error);
     }

apps/api/src/services/DomainService.ts

@@ -89,10 +89,13 @@ export class DomainService {
 
       // Send email notification about domain verified
       try {
-        // Check deduplication cache
+        // Use SETNX to atomically check and set the flag (prevents race conditions)
         const cacheKey = Keys.Domain.verifiedEmail(domainId);
-        const alreadySent = await redis.get(cacheKey);
+        const ttl = 604800; // 7 days
-        if (alreadySent !== '1') {
+
+        // SETNX returns 1 if key was set (didn't exist), 0 if key already existed
+        const wasSet = await redis.set(cacheKey, '1', 'EX', ttl, 'NX');
+        if (wasSet) {
           const members = await MembershipService.getMembers(updatedDomain.project.id);
           const emails = members.map(m => m.email);
           if (emails.length > 0) {
@@ -104,8 +107,6 @@ export class DomainService {
               landingUrl: LANDING_URI,
             });
             await Promise.all(emails.map(email => sendPlatformEmail(email, 'Domain Verified Successfully', template)));
-            // Set cache to prevent duplicate emails (7 days)
-            await redis.setex(cacheKey, 604800, '1');
           }
         }
       } catch (emailError) {
@@ -131,13 +132,17 @@ export class DomainService {
 
       // Send email notification about domain verification failed
       try {
-        // Check deduplication cache (monthly)
+        // Use SETNX to atomically check and set the flag (prevents race conditions)
         const now = new Date();
         const year = now.getFullYear();
         const month = String(now.getMonth() + 1).padStart(2, '0');
         const cacheKey = Keys.Domain.unverifiedEmail(domainId, year, month);
-        const alreadySent = await redis.get(cacheKey);
+        const endOfMonth = new Date(now.getFullYear(), now.getMonth() + 1, 1);
-        if (alreadySent !== '1') {
+        const ttl = Math.floor((endOfMonth.getTime() - now.getTime()) / 1000);
+
+        // SETNX returns 1 if key was set (didn't exist), 0 if key already existed
+        const wasSet = await redis.set(cacheKey, '1', 'EX', ttl, 'NX');
+        if (wasSet) {
           const members = await MembershipService.getMembers(updatedDomain.project.id);
           const emails = members.map(m => m.email);
           if (emails.length > 0) {
@@ -149,10 +154,6 @@ export class DomainService {
               landingUrl: LANDING_URI,
             });
             await Promise.all(emails.map(email => sendPlatformEmail(email, 'Domain Verification Failed', template)));
-            // Set cache to prevent duplicate emails (until end of month)
-            const endOfMonth = new Date(now.getFullYear(), now.getMonth() + 1, 1);
-            const ttl = Math.floor((endOfMonth.getTime() - now.getTime()) / 1000);
-            await redis.setex(cacheKey, ttl, '1');
           }
         }
       } catch (emailError) {

apps/api/src/services/NtfyService.ts

@@ -211,9 +211,11 @@ export class NtfyService {
     const {redis} = await import('../database/redis.js');
 
     const cacheKey = `ntfy:security:warning:${projectId}`;
-    const exists = await redis.exists(cacheKey);
+    const ttl = 3600; // 1 hour
 
-    if (exists) {
+    // Use SETNX to atomically check and set the flag (prevents race conditions)
+    const wasSet = await redis.set(cacheKey, '1', 'EX', ttl, 'NX');
+    if (!wasSet) {
       return;
     }
 
@@ -223,9 +225,6 @@ export class NtfyService {
       `Project "${projectName}" (${projectId}) has security warnings: ${warningText}`,
       [NtfyTag.WARNING, NtfyTag.SHIELD],
     );
-
-    // Throttle for 1 hour
-    await redis.setex(cacheKey, 3600, '1');
   }
 
   /**
@@ -606,9 +605,11 @@ export class NtfyService {
     const {redis} = await import('../database/redis.js');
 
     const cacheKey = `ntfy:billing:warning:${projectId}:${sourceType}`;
-    const exists = await redis.exists(cacheKey);
+    const ttl = 86400; // 24 hours
 
-    if (exists) {
+    // Use SETNX to atomically check and set the flag (prevents race conditions)
+    const wasSet = await redis.set(cacheKey, '1', 'EX', ttl, 'NX');
+    if (!wasSet) {
       return;
     }
 
@@ -617,9 +618,6 @@ export class NtfyService {
       `Email usage at ${Math.round(percentage)}% (${usage}/${limit}) for ${sourceType} in project "${projectName}" (${projectId})`,
       [NtfyTag.WARNING, NtfyTag.MONEY, NtfyTag.CHART],
     );
-
-    // Throttle for 24 hours
-    await redis.setex(cacheKey, 86400, '1');
   }
 
   // ===== Billing and usage limit notifications =====
@@ -638,9 +636,11 @@ export class NtfyService {
     const {redis} = await import('../database/redis.js');
 
     const cacheKey = `ntfy:billing:exceeded:${projectId}:${sourceType}`;
-    const exists = await redis.exists(cacheKey);
+    const ttl = 86400; // 24 hours
 
-    if (exists) {
+    // Use SETNX to atomically check and set the flag (prevents race conditions)
+    const wasSet = await redis.set(cacheKey, '1', 'EX', ttl, 'NX');
+    if (!wasSet) {
       return;
     }
 
@@ -649,9 +649,6 @@ export class NtfyService {
       `Email usage limit reached (${usage}/${limit}) for ${sourceType} in project "${projectName}" (${projectId}). Further emails are blocked.`,
       [NtfyTag.ERROR, NtfyTag.MONEY, NtfyTag.SKULL],
     );
-
-    // Throttle for 24 hours
-    await redis.setex(cacheKey, 86400, '1');
   }
 
   /**