-
Notifications
You must be signed in to change notification settings - Fork 194
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
187 changed files
with
2,031 additions
and
3,119 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,6 @@ | ||
| export class RateLimitError extends Error { | ||
| constructor(message: string, public retryAfter: number) { | ||
| super(message); | ||
| this.name = 'RateLimitError'; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,33 @@ | ||
| import { Queues } from '@@core/@core-services/queues/types'; | ||
| import { IngestDataService } from '@@core/@core-services/unification/ingest-data.service'; | ||
| import { Process, Processor } from '@nestjs/bull'; | ||
| import { Job } from 'bull'; | ||
|
|
||
| @Processor(Queues.RATE_LIMIT_FAILED_JOBS) | ||
| export class RateLimitJobProcessor { | ||
| constructor(private ingestDataService: IngestDataService) {} | ||
|
|
||
| @Process('rate-limit-sync') | ||
| async processRateLimitedJob(job: Job<{ method: string; args: any[] }>) { | ||
| const { method, args } = job.data; | ||
| try { | ||
| if (method === 'syncForLinkedUser') { | ||
| await this.ingestDataService.syncForLinkedUser( | ||
| ...(args as Parameters< | ||
| typeof this.ingestDataService.syncForLinkedUser | ||
| >), | ||
| ); | ||
| } | ||
|
|
||
| // Fallback for other methods (if any) | ||
| /*const targetInstance = this.moduleRef.get(target, { strict: false }); | ||
| if (targetInstance && typeof targetInstance[method] === 'function') { | ||
| await targetInstance[method](...args); | ||
| return; | ||
| }*/ | ||
| } catch (error) { | ||
| console.error(`Error processing rate-limited job: ${error.message}`); | ||
| throw error; | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,33 @@ | ||
| import { RateLimitService } from './rate-limit.service'; | ||
|
|
||
| export function RateLimit() { | ||
| return function ( | ||
| target: any, | ||
| propertyKey: string, | ||
| descriptor: PropertyDescriptor, | ||
| ) { | ||
| const originalMethod = descriptor.value; | ||
|
|
||
| descriptor.value = async function (...args: any[]) { | ||
| const rateLimitService: RateLimitService = (this as any).rateLimitService; | ||
| const { connection } = args[0]; | ||
|
|
||
| if (!rateLimitService) { | ||
| console.error('RateLimitService not found in the class instance'); | ||
| return originalMethod.apply(this, args); | ||
| } | ||
|
|
||
| try { | ||
| await rateLimitService.checkRateLimit( | ||
| connection.id_connection, | ||
| connection.provider_slug, | ||
| ); | ||
| return await originalMethod.apply(this, args); | ||
| } catch (error) { | ||
| throw error; | ||
| } | ||
| }; | ||
|
|
||
| return descriptor; | ||
| }; | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,9 @@ | ||
| import { Module } from '@nestjs/common'; | ||
| import { RateLimitJobProcessor } from './rate-limit.consumer'; | ||
| import { PrismaService } from '@@core/@core-services/prisma/prisma.service'; | ||
| import { WebhookService } from '@@core/@core-services/webhooks/panora-webhooks/webhook.service'; | ||
|
|
||
| @Module({ | ||
| providers: [RateLimitJobProcessor, WebhookService, PrismaService], | ||
| }) | ||
| export class RateLimitModule {} |
137 changes: 137 additions & 0 deletions
137
packages/api/src/@core/rate-limit/rate-limit.service.ts
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,137 @@ | ||
| import { PrismaService } from '@@core/@core-services/prisma/prisma.service'; | ||
| import { Injectable } from '@nestjs/common'; | ||
| import { RateLimitError } from './error'; | ||
|
|
||
| interface RateLimitPolicy { | ||
| timeWindow: number; | ||
| maxRequests: number; | ||
| } | ||
|
|
||
| @Injectable() | ||
| export class RateLimitService { | ||
| constructor(private prisma: PrismaService) {} | ||
|
|
||
| async checkRateLimit( | ||
| connectionId: string, | ||
| providerSlug: string, | ||
| ): Promise<boolean> { | ||
| const policies = await this.getRateLimitPolicies(providerSlug); | ||
|
|
||
| for (const policy of policies) { | ||
| const { timeWindow, maxRequests } = policy; | ||
| const windowStart = new Date(Date.now() - timeWindow * 1000); | ||
|
|
||
| const requestCount = await this.prisma.events.count({ | ||
| where: { | ||
| id_connection: connectionId, | ||
| timestamp: { gte: windowStart }, | ||
| type: { endsWith: '.pulled' }, | ||
| }, | ||
| }); | ||
|
|
||
| if (requestCount >= maxRequests) { | ||
| const retryAfter = await this.getRetryAfter(connectionId); | ||
| throw new RateLimitError('Rate limit exceeded', retryAfter); | ||
| } | ||
| } | ||
|
|
||
| return true; // All checks passed | ||
| } | ||
|
|
||
| private async getRateLimitPolicies( | ||
| providerSlug: string, | ||
| ): Promise<RateLimitPolicy[]> { | ||
| const policies: Record<string, RateLimitPolicy[]> = { | ||
| hubspot: [ | ||
| { timeWindow: 10, maxRequests: 110 }, // 110 calls per 10 seconds | ||
| { timeWindow: 86400, maxRequests: 250000 }, // 250k calls per day | ||
| ], | ||
| }; | ||
| return policies[providerSlug] || []; | ||
| } | ||
|
|
||
| async getRetryAfter(connectionId: string): Promise<number> { | ||
| const connection = await this.prisma.connections.findUnique({ | ||
| where: { id_connection: connectionId }, | ||
| }); | ||
|
|
||
| if (!connection) { | ||
| throw new Error(`Connection not found for id: ${connectionId}`); | ||
| } | ||
|
|
||
| const policies = await this.getRateLimitPolicies(connection.provider_slug); | ||
|
|
||
| if (policies.length === 0) { | ||
| return 10000; // 10 seconds default delay if no policies | ||
| } | ||
|
|
||
| let maxTimeUntilReset = 0; | ||
|
|
||
| for (const policy of policies) { | ||
| const windowStart = new Date(Date.now() - policy.timeWindow * 1000); | ||
| const requestCount = await this.prisma.events.count({ | ||
| where: { | ||
| id_connection: connectionId, | ||
| timestamp: { gte: windowStart }, | ||
| type: { endsWith: '.pulled' }, | ||
| }, | ||
| }); | ||
|
|
||
| if (requestCount >= policy.maxRequests) { | ||
| const latestEvent = await this.prisma.events.findFirst({ | ||
| where: { | ||
| id_connection: connectionId, | ||
| type: { endsWith: '.pulled' }, | ||
| }, | ||
| orderBy: { timestamp: 'desc' }, | ||
| }); | ||
|
|
||
| if (latestEvent) { | ||
| const timeSinceLastRequest = | ||
| Date.now() - latestEvent.timestamp.getTime(); | ||
| const timeUntilReset = | ||
| policy.timeWindow * 1000 - timeSinceLastRequest; | ||
| maxTimeUntilReset = Math.max(maxTimeUntilReset, timeUntilReset); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| if (maxTimeUntilReset <= 0) { | ||
| return 0; | ||
| } | ||
|
|
||
| const buffer = 1000; // 1 second buffer | ||
| return maxTimeUntilReset + buffer; | ||
| } | ||
|
|
||
| /*async retryWithBackoff(config: any): Promise<AxiosResponse> { | ||
| return backOff( | ||
| async () => { | ||
| try { | ||
| const response = await axios(config); | ||
| return response; | ||
| } catch (error) { | ||
| if (error.response && error.response.status === 429) { | ||
| const retryAfter = await this.getRetryAfter(config.connectionId); | ||
| if (retryAfter) { | ||
| await new Promise((resolve) => setTimeout(resolve, retryAfter)); | ||
| } | ||
| throw error; // Rethrow to trigger backoff | ||
| } | ||
| throw error; // Rethrow non-rate-limit errors | ||
| } | ||
| }, | ||
| { | ||
| numOfAttempts: 10, | ||
| startingDelay: 1000, | ||
| timeMultiple: 2, | ||
| maxDelay: 60000, | ||
| jitter: 'full', | ||
| retry: (e: Error, attemptNumber: number) => { | ||
| console.log(`Retry attempt ${attemptNumber} due to: ${e.message}`); | ||
| return true; | ||
| }, | ||
| }, | ||
| ); | ||
| }*/ | ||
| } |
Oops, something went wrong.