feat(redis): added redis option for rate limiter, 10x speed improvement in rate limit checks & reduction of DB load (#2263)

* feat(redis): added redis option for rate limiter, 10x speed improvement in rate limit checks & reduction of DB load

* ack PR comments

* improvements

Author: waleedlatif1

apps/sim/app/api/users/me/usage-limits/route.ts

@@ -4,9 +4,9 @@ import { checkServerSideUsageLimits } from '@/lib/billing'
 import { getHighestPrioritySubscription } from '@/lib/billing/core/subscription'
 import { getEffectiveCurrentPeriodCost } from '@/lib/billing/core/usage'
 import { getUserStorageLimit, getUserStorageUsage } from '@/lib/billing/storage'
+import { RateLimiter } from '@/lib/core/rate-limiter'
 import { createLogger } from '@/lib/logs/console/logger'
 import { createErrorResponse } from '@/app/api/workflows/utils'
-import { RateLimiter } from '@/services/queue'
 
 const logger = createLogger('UsageLimitsAPI')
 

apps/sim/app/api/v1/logs/meta.ts

@@ -1,7 +1,7 @@
 import { checkServerSideUsageLimits } from '@/lib/billing'
 import { getHighestPrioritySubscription } from '@/lib/billing/core/subscription'
 import { getEffectiveCurrentPeriodCost } from '@/lib/billing/core/usage'
-import { RateLimiter } from '@/services/queue'
+import { RateLimiter } from '@/lib/core/rate-limiter'
 
 export interface UserLimits {
   workflowExecutionRateLimit: {

apps/sim/app/api/v1/middleware.ts

@@ -1,8 +1,8 @@
 import { type NextRequest, NextResponse } from 'next/server'
 import { getHighestPrioritySubscription } from '@/lib/billing/core/subscription'
+import { RateLimiter } from '@/lib/core/rate-limiter/rate-limiter'
 import { createLogger } from '@/lib/logs/console/logger'
 import { authenticateV1Request } from '@/app/api/v1/auth'
-import { RateLimiter } from '@/services/queue/RateLimiter'
 
 const logger = createLogger('V1Middleware')
 const rateLimiter = new RateLimiter()

apps/sim/app/api/webhooks/trigger/[path]/route.test.ts

@@ -140,7 +140,7 @@ vi.mock('@/lib/workspaces/utils', async () => {
   }
 })
 
-vi.mock('@/services/queue', () => ({
+vi.mock('@/lib/core/rate-limiter', () => ({
   RateLimiter: vi.fn().mockImplementation(() => ({
     checkRateLimit: vi.fn().mockResolvedValue({
       allowed: true,

apps/sim/app/api/workflows/[id]/execute/route.ts

@@ -395,8 +395,7 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id:
       triggerType: loggingTriggerType,
       executionId,
       requestId,
-      checkRateLimit: false, // Manual executions bypass rate limits
-      checkDeployment: !shouldUseDraftState, // Check deployment unless using draft
+      checkDeployment: !shouldUseDraftState,
       loggingSession,
     })
 

apps/sim/background/workspace-notification-delivery.ts

@@ -11,13 +11,13 @@ import { and, eq, isNull, lte, or, sql } from 'drizzle-orm'
 import { v4 as uuidv4 } from 'uuid'
 import { checkUsageStatus } from '@/lib/billing/calculations/usage-monitor'
 import { getHighestPrioritySubscription } from '@/lib/billing/core/subscription'
+import { RateLimiter } from '@/lib/core/rate-limiter'
 import { decryptSecret } from '@/lib/core/security/encryption'
 import { getBaseUrl } from '@/lib/core/utils/urls'
 import { createLogger } from '@/lib/logs/console/logger'
 import type { TraceSpan, WorkflowExecutionLog } from '@/lib/logs/types'
 import { sendEmail } from '@/lib/messaging/email/mailer'
 import type { AlertConfig } from '@/lib/notifications/alert-rules'
-import { RateLimiter } from '@/services/queue'
 
 const logger = createLogger('WorkspaceNotificationDelivery')
 

apps/sim/lib/core/rate-limiter/index.ts

@@ -0,0 +1,7 @@
+export { RateLimiter } from '@/lib/core/rate-limiter/rate-limiter'
+export type {
+  RateLimitConfig,
+  SubscriptionPlan,
+  TriggerType,
+} from '@/lib/core/rate-limiter/types'
+export { RATE_LIMITS, RateLimitError } from '@/lib/core/rate-limiter/types'

apps/sim/lib/core/rate-limiter/rate-limiter.test.ts

@@ -0,0 +1,309 @@
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+import { RateLimiter } from '@/lib/core/rate-limiter/rate-limiter'
+import { MANUAL_EXECUTION_LIMIT, RATE_LIMITS } from '@/lib/core/rate-limiter/types'
+
+vi.mock('@sim/db', () => ({
+  db: {
+    select: vi.fn(),
+    insert: vi.fn(),
+    update: vi.fn(),
+    delete: vi.fn(),
+  },
+}))
+
+vi.mock('drizzle-orm', () => ({
+  eq: vi.fn((field, value) => ({ field, value })),
+  sql: vi.fn((strings, ...values) => ({ sql: strings.join('?'), values })),
+  and: vi.fn((...conditions) => ({ and: conditions })),
+}))
+
+vi.mock('@/lib/core/config/redis', () => ({
+  getRedisClient: vi.fn().mockReturnValue(null),
+}))
+
+import { db } from '@sim/db'
+import { getRedisClient } from '@/lib/core/config/redis'
+
+describe('RateLimiter', () => {
+  const rateLimiter = new RateLimiter()
+  const testUserId = 'test-user-123'
+  const freeSubscription = { plan: 'free', referenceId: testUserId }
+
+  beforeEach(() => {
+    vi.clearAllMocks()
+    vi.mocked(getRedisClient).mockReturnValue(null)
+  })
+
+  describe('checkRateLimitWithSubscription', () => {
+    it('should allow unlimited requests for manual trigger type', async () => {
+      const result = await rateLimiter.checkRateLimitWithSubscription(
+        testUserId,
+        freeSubscription,
+        'manual',
+        false
+      )
+
+      expect(result.allowed).toBe(true)
+      expect(result.remaining).toBe(MANUAL_EXECUTION_LIMIT)
+      expect(result.resetAt).toBeInstanceOf(Date)
+      expect(db.select).not.toHaveBeenCalled()
+    })
+
+    it('should allow first API request for sync execution (DB fallback)', async () => {
+      vi.mocked(db.select).mockReturnValue({
+        from: vi.fn().mockReturnValue({
+          where: vi.fn().mockReturnValue({
+            limit: vi.fn().mockResolvedValue([]),
+          }),
+        }),
+      } as any)
+
+      vi.mocked(db.insert).mockReturnValue({
+        values: vi.fn().mockReturnValue({
+          onConflictDoUpdate: vi.fn().mockReturnValue({
+            returning: vi.fn().mockResolvedValue([
+              {
+                syncApiRequests: 1,
+                asyncApiRequests: 0,
+                apiEndpointRequests: 0,
+                windowStart: new Date(),
+              },
+            ]),
+          }),
+        }),
+      } as any)
+
+      const result = await rateLimiter.checkRateLimitWithSubscription(
+        testUserId,
+        freeSubscription,
+        'api',
+        false
+      )
+
+      expect(result.allowed).toBe(true)
+      expect(result.remaining).toBe(RATE_LIMITS.free.syncApiExecutionsPerMinute - 1)
+      expect(result.resetAt).toBeInstanceOf(Date)
+    })
+
+    it('should allow first API request for async execution (DB fallback)', async () => {
+      vi.mocked(db.select).mockReturnValue({
+        from: vi.fn().mockReturnValue({
+          where: vi.fn().mockReturnValue({
+            limit: vi.fn().mockResolvedValue([]),
+          }),
+        }),
+      } as any)
+
+      vi.mocked(db.insert).mockReturnValue({
+        values: vi.fn().mockReturnValue({
+          onConflictDoUpdate: vi.fn().mockReturnValue({
+            returning: vi.fn().mockResolvedValue([
+              {
+                syncApiRequests: 0,
+                asyncApiRequests: 1,
+                apiEndpointRequests: 0,
+                windowStart: new Date(),
+              },
+            ]),
+          }),
+        }),
+      } as any)
+
+      const result = await rateLimiter.checkRateLimitWithSubscription(
+        testUserId,
+        freeSubscription,
+        'api',
+        true
+      )
+
+      expect(result.allowed).toBe(true)
+      expect(result.remaining).toBe(RATE_LIMITS.free.asyncApiExecutionsPerMinute - 1)
+      expect(result.resetAt).toBeInstanceOf(Date)
+    })
+
+    it('should work for all trigger types except manual (DB fallback)', async () => {
+      const triggerTypes = ['api', 'webhook', 'schedule', 'chat'] as const
+
+      for (const triggerType of triggerTypes) {
+        vi.mocked(db.select).mockReturnValue({
+          from: vi.fn().mockReturnValue({
+            where: vi.fn().mockReturnValue({
+              limit: vi.fn().mockResolvedValue([]),
+            }),
+          }),
+        } as any)
+
+        vi.mocked(db.insert).mockReturnValue({
+          values: vi.fn().mockReturnValue({
+            onConflictDoUpdate: vi.fn().mockReturnValue({
+              returning: vi.fn().mockResolvedValue([
+                {
+                  syncApiRequests: 1,
+                  asyncApiRequests: 0,
+                  apiEndpointRequests: 0,
+                  windowStart: new Date(),
+                },
+              ]),
+            }),
+          }),
+        } as any)
+
+        const result = await rateLimiter.checkRateLimitWithSubscription(
+          testUserId,
+          freeSubscription,
+          triggerType,
+          false
+        )
+
+        expect(result.allowed).toBe(true)
+        expect(result.remaining).toBe(RATE_LIMITS.free.syncApiExecutionsPerMinute - 1)
+      }
+    })
+
+    it('should use Redis when available', async () => {
+      const mockRedis = {
+        eval: vi.fn().mockResolvedValue(1), // Lua script returns count after INCR
+      }
+      vi.mocked(getRedisClient).mockReturnValue(mockRedis as any)
+
+      const result = await rateLimiter.checkRateLimitWithSubscription(
+        testUserId,
+        freeSubscription,
+        'api',
+        false
+      )
+
+      expect(result.allowed).toBe(true)
+      expect(result.remaining).toBe(RATE_LIMITS.free.syncApiExecutionsPerMinute - 1)
+      expect(mockRedis.eval).toHaveBeenCalled()
+      expect(db.select).not.toHaveBeenCalled()
+    })
+
+    it('should deny requests when Redis rate limit exceeded', async () => {
+      const mockRedis = {
+        eval: vi.fn().mockResolvedValue(RATE_LIMITS.free.syncApiExecutionsPerMinute + 1),
+      }
+      vi.mocked(getRedisClient).mockReturnValue(mockRedis as any)
+
+      const result = await rateLimiter.checkRateLimitWithSubscription(
+        testUserId,
+        freeSubscription,
+        'api',
+        false
+      )
+
+      expect(result.allowed).toBe(false)
+      expect(result.remaining).toBe(0)
+    })
+
+    it('should fall back to DB when Redis fails', async () => {
+      const mockRedis = {
+        eval: vi.fn().mockRejectedValue(new Error('Redis connection failed')),
+      }
+      vi.mocked(getRedisClient).mockReturnValue(mockRedis as any)
+
+      vi.mocked(db.select).mockReturnValue({
+        from: vi.fn().mockReturnValue({
+          where: vi.fn().mockReturnValue({
+            limit: vi.fn().mockResolvedValue([]),
+          }),
+        }),
+      } as any)
+
+      vi.mocked(db.insert).mockReturnValue({
+        values: vi.fn().mockReturnValue({
+          onConflictDoUpdate: vi.fn().mockReturnValue({
+            returning: vi.fn().mockResolvedValue([
+              {
+                syncApiRequests: 1,
+                asyncApiRequests: 0,
+                apiEndpointRequests: 0,
+                windowStart: new Date(),
+              },
+            ]),
+          }),
+        }),
+      } as any)
+
+      const result = await rateLimiter.checkRateLimitWithSubscription(
+        testUserId,
+        freeSubscription,
+        'api',
+        false
+      )
+
+      expect(result.allowed).toBe(true)
+      expect(db.select).toHaveBeenCalled()
+    })
+  })
+
+  describe('getRateLimitStatusWithSubscription', () => {
+    it('should return unlimited for manual trigger type', async () => {
+      const status = await rateLimiter.getRateLimitStatusWithSubscription(
+        testUserId,
+        freeSubscription,
+        'manual',
+        false
+      )
+
+      expect(status.used).toBe(0)
+      expect(status.limit).toBe(MANUAL_EXECUTION_LIMIT)
+      expect(status.remaining).toBe(MANUAL_EXECUTION_LIMIT)
+      expect(status.resetAt).toBeInstanceOf(Date)
+    })
+
+    it('should return sync API limits for API trigger type (DB fallback)', async () => {
+      vi.mocked(db.select).mockReturnValue({
+        from: vi.fn().mockReturnValue({
+          where: vi.fn().mockReturnValue({
+            limit: vi.fn().mockResolvedValue([]),
+          }),
+        }),
+      } as any)
+
+      const status = await rateLimiter.getRateLimitStatusWithSubscription(
+        testUserId,
+        freeSubscription,
+        'api',
+        false
+      )
+
+      expect(status.used).toBe(0)
+      expect(status.limit).toBe(RATE_LIMITS.free.syncApiExecutionsPerMinute)
+      expect(status.remaining).toBe(RATE_LIMITS.free.syncApiExecutionsPerMinute)
+      expect(status.resetAt).toBeInstanceOf(Date)
+    })
+
+    it('should use Redis for status when available', async () => {
+      const mockRedis = {
+        get: vi.fn().mockResolvedValue('5'),
+      }
+      vi.mocked(getRedisClient).mockReturnValue(mockRedis as any)
+
+      const status = await rateLimiter.getRateLimitStatusWithSubscription(
+        testUserId,
+        freeSubscription,
+        'api',
+        false
+      )
+
+      expect(status.used).toBe(5)
+      expect(status.limit).toBe(RATE_LIMITS.free.syncApiExecutionsPerMinute)
+      expect(status.remaining).toBe(RATE_LIMITS.free.syncApiExecutionsPerMinute - 5)
+      expect(mockRedis.get).toHaveBeenCalled()
+      expect(db.select).not.toHaveBeenCalled()
+    })
+  })
+
+  describe('resetRateLimit', () => {
+    it('should delete rate limit record for user', async () => {
+      vi.mocked(db.delete).mockReturnValue({
+        where: vi.fn().mockResolvedValue({}),
+      } as any)
+
+      await rateLimiter.resetRateLimit(testUserId)
+
+      expect(db.delete).toHaveBeenCalled()
+    })
+  })
+})