From 4db1ccc3fc59edb0187d8e2223f45bb0bce6deb5 Mon Sep 17 00:00:00 2001 From: Dries Augustyns Date: Mon, 12 Jan 2026 08:33:51 +0100 Subject: [PATCH] fix: Refactor Redis keys to prevent multiple messages on concurrent requests --- apps/api/src/jobs/domain-verification.ts | 21 ++++++++----- apps/api/src/services/BillingLimitService.ts | 32 +++++++++++--------- apps/api/src/services/DomainService.ts | 25 +++++++-------- apps/api/src/services/NtfyService.ts | 27 ++++++++--------- 4 files changed, 56 insertions(+), 49 deletions(-) diff --git a/apps/api/src/jobs/domain-verification.ts b/apps/api/src/jobs/domain-verification.ts index b3a2e13..f8d2cc2 100644 --- a/apps/api/src/jobs/domain-verification.ts +++ b/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); - if (alreadySent !== '1') { + const ttl = 604800; // 7 days + + // 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); - if (alreadySent !== '1') { + 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) { 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) { diff --git a/apps/api/src/services/BillingLimitService.ts b/apps/api/src/services/BillingLimitService.ts index ad0bc90..fe95dc3 100644 --- a/apps/api/src/services/BillingLimitService.ts +++ b/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); - if (alreadySent === '1') { + // Use SETNX (SET if Not eXists) to atomically check and set the flag + // 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); - if (alreadySent === '1') { + // Use SETNX (SET if Not eXists) to atomically check and set the flag + // 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); } diff --git a/apps/api/src/services/DomainService.ts b/apps/api/src/services/DomainService.ts index 2e54f2b..4e07c9f 100644 --- a/apps/api/src/services/DomainService.ts +++ b/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); - if (alreadySent !== '1') { + const ttl = 604800; // 7 days + + // 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); - if (alreadySent !== '1') { + 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) { 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) { diff --git a/apps/api/src/services/NtfyService.ts b/apps/api/src/services/NtfyService.ts index 37508d2..ccafa15 100644 --- a/apps/api/src/services/NtfyService.ts +++ b/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'); } /**