feat: add global SQS configuration #1893
Closed
Conversation
This file contains hidden or 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
Reviewer's GuideIntroduces global SQS configuration, enabling prefix-based queue creation and event filtering, enforces payload size limits with optional S3 fallback, and refactors controller methods for uniform queue management under global or per-instance modes. Sequence diagram for SQS event dispatch with global configuration and S3 fallbacksequenceDiagram
participant Controller as SqsController
participant Config as ConfigService
participant SQS as SQS
participant S3 as S3Service
participant Logger as Logger
Controller->>Config: get SQS config
alt GLOBAL_ENABLED
Controller->>SQS: sendMessage (global queue)
else per-instance
Controller->>SQS: sendMessage (instance queue)
end
Controller->>Config: get MAX_PAYLOAD_SIZE
Controller->>SQS: prepare message
alt payload size > MAX_PAYLOAD_SIZE
Controller->>Config: get S3 config
alt S3 ENABLED
Controller->>S3: uploadFile
S3-->>Controller: fileUrl
Controller->>SQS: sendMessage (with fileUrl)
else S3 not enabled
Controller->>Logger: error (payload too large)
end
else payload size OK
Controller->>SQS: sendMessage (with message)
end
Class diagram for updated SqsController and Sqs config typesclassDiagram
class SqsController {
+sqs: SQS
+logger
+monitor
+name
+status
+set(instanceName: string, data: EventDto): Promise<any>
+saveQueues(prefixName: string, events: string[], enable: boolean)
+listQueues(prefixName: string)
+removeQueuesByInstance(prefixName: string)
}
class Sqs {
+ENABLED: boolean
+GLOBAL_ENABLED: boolean
+GLOBAL_PREFIX_NAME: string
+ACCESS_KEY_ID: string
+SECRET_ACCESS_KEY: string
+ACCOUNT_ID: string
+REGION: string
+MAX_PAYLOAD_SIZE: number
+EVENTS: object
}
SqsController --> Sqs : uses
class ConfigService {
+envProcess(): Env
}
ConfigService --> Sqs : returns Sqs config
class HttpServer {
+NAME: string
+TYPE: 'http' | 'https'
+PORT: number
+URL: string
}
SqsController --> HttpServer : uses
class S3 {
+ENABLE: boolean
}
SqsController --> S3 : uses
File-Level Changes
Possibly linked issues
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hey there - I've reviewed your changes and they look great!
Blocking issues:
- Detected possible user input going into a
path.joinorpath.resolvefunction. This could possibly lead to a path traversal vulnerability, where the attacker can access arbitrary files stored in the file system. Instead, be sure to sanitize or validate user input first. (link)
Prompt for AI Agents
Please address the comments from this code review:
## Individual Comments
### Comment 1
<location> `src/api/integrations/event/sqs/sqs.controller.ts:25` </location>
<code_context>
}
- new Promise<void>((resolve) => {
+ new Promise<void>(async (resolve) => {
const awsConfig = configService.get<Sqs>('SQS');
</code_context>
<issue_to_address>
Using an async executor in a Promise is discouraged.
Refactor to avoid passing an async function to the Promise constructor, as this may cause unhandled rejections. Use an async function directly or restructure the logic.
</issue_to_address>
### Comment 2
<location> `src/config/env.config.ts:508` </location>
<code_context>
SECRET_ACCESS_KEY: process.env.SQS_SECRET_ACCESS_KEY || '',
ACCOUNT_ID: process.env.SQS_ACCOUNT_ID || '',
REGION: process.env.SQS_REGION || '',
+ MAX_PAYLOAD_SIZE: Number.parseInt(process.env.SQS_MAX_PAYLOAD_SIZE) || 1048576,
+ EVENTS: {
+ APPLICATION_STARTUP: process.env?.SQS_GLOBAL_APPLICATION_STARTUP === 'true',
</code_context>
<issue_to_address>
Default MAX_PAYLOAD_SIZE may mask misconfiguration.
Log a warning if SQS_MAX_PAYLOAD_SIZE is missing or invalid to help detect configuration issues early.
</issue_to_address>
<suggested_fix>
<<<<<<< SEARCH
MAX_PAYLOAD_SIZE: Number.parseInt(process.env.SQS_MAX_PAYLOAD_SIZE) || 1048576,
=======
MAX_PAYLOAD_SIZE: (() => {
const raw = process.env.SQS_MAX_PAYLOAD_SIZE;
const parsed = Number.parseInt(raw ?? '');
if (!raw || isNaN(parsed)) {
console.warn(
'[config] Warning: SQS_MAX_PAYLOAD_SIZE is missing or invalid. Using default value 1048576.'
);
return 1048576;
}
return parsed;
})(),
>>>>>>> REPLACE
</suggested_fix>
## Security Issues
### Issue 1
<location> `src/api/integrations/event/sqs/sqs.controller.ts:156` </location>
<issue_to_address>
**security (javascript.lang.security.audit.path-traversal.path-join-resolve-traversal):** Detected possible user input going into a `path.join` or `path.resolve` function. This could possibly lead to a path traversal vulnerability, where the attacker can access arbitrary files stored in the file system. Instead, be sure to sanitize or validate user input first.
*Source: opengrep*
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
| } | ||
|
|
||
| new Promise<void>((resolve) => { | ||
| new Promise<void>(async (resolve) => { |
There was a problem hiding this comment.
issue: Using an async executor in a Promise is discouraged.
Refactor to avoid passing an async function to the Promise constructor, as this may cause unhandled rejections. Use an async function directly or restructure the logic.
src/config/env.config.ts
Outdated
| SECRET_ACCESS_KEY: process.env.SQS_SECRET_ACCESS_KEY || '', | ||
| ACCOUNT_ID: process.env.SQS_ACCOUNT_ID || '', | ||
| REGION: process.env.SQS_REGION || '', | ||
| MAX_PAYLOAD_SIZE: Number.parseInt(process.env.SQS_MAX_PAYLOAD_SIZE) || 1048576, |
There was a problem hiding this comment.
suggestion: Default MAX_PAYLOAD_SIZE may mask misconfiguration.
Log a warning if SQS_MAX_PAYLOAD_SIZE is missing or invalid to help detect configuration issues early.
Suggested change
| MAX_PAYLOAD_SIZE: Number.parseInt(process.env.SQS_MAX_PAYLOAD_SIZE) || 1048576, | |
| MAX_PAYLOAD_SIZE: (() => { | |
| const raw = process.env.SQS_MAX_PAYLOAD_SIZE; | |
| const parsed = Number.parseInt(raw ?? ''); | |
| if (!raw || isNaN(parsed)) { | |
| console.warn( | |
| '[config] Warning: SQS_MAX_PAYLOAD_SIZE is missing or invalid. Using default value 1048576.' | |
| ); | |
| return 1048576; | |
| } | |
| return parsed; | |
| })(), |
| const fileName = `${instanceName}_${eventFormatted}_${Date.now()}.json`; | ||
| const fullName = join( | ||
| 'messages', | ||
| fileName |
There was a problem hiding this comment.
security (javascript.lang.security.audit.path-traversal.path-join-resolve-traversal): Detected possible user input going into a path.join or path.resolve function. This could possibly lead to a path traversal vulnerability, where the attacker can access arbitrary files stored in the file system. Instead, be sure to sanitize or validate user input first.
Source: opengrep
Comment on lines
181
to
203
| if (err) { | ||
| this.logger.error({ | ||
| local: `${origin}.sendData-SQS`, | ||
| params: JSON.stringify(message), | ||
| sqsUrl: sqsUrl, | ||
| message: err?.message, | ||
| hostName: err?.hostname, | ||
| code: err?.code, | ||
| stack: err?.stack, | ||
| name: err?.name, | ||
| url: queueName, | ||
| server_url: serverUrl, | ||
| }); | ||
| } else { | ||
| if (configService.get<Log>('LOG').LEVEL.includes('WEBHOOKS')) { | ||
| const logData = { | ||
| local: `${origin}.sendData-SQS`, | ||
| ...message, | ||
| }; | ||
|
|
||
| this.logger.log(logData); | ||
| } | ||
| } |
There was a problem hiding this comment.
suggestion (code-quality): Merge else clause's nested if statement into else if (merge-else-if)
Suggested change
| if (err) { | |
| this.logger.error({ | |
| local: `${origin}.sendData-SQS`, | |
| params: JSON.stringify(message), | |
| sqsUrl: sqsUrl, | |
| message: err?.message, | |
| hostName: err?.hostname, | |
| code: err?.code, | |
| stack: err?.stack, | |
| name: err?.name, | |
| url: queueName, | |
| server_url: serverUrl, | |
| }); | |
| } else { | |
| if (configService.get<Log>('LOG').LEVEL.includes('WEBHOOKS')) { | |
| const logData = { | |
| local: `${origin}.sendData-SQS`, | |
| ...message, | |
| }; | |
| this.logger.log(logData); | |
| } | |
| } | |
| if (err) { | |
| this.logger.error({ | |
| local: `${origin}.sendData-SQS`, | |
| params: JSON.stringify(message), | |
| sqsUrl: sqsUrl, | |
| message: err?.message, | |
| hostName: err?.hostname, | |
| code: err?.code, | |
| stack: err?.stack, | |
| name: err?.name, | |
| url: queueName, | |
| server_url: serverUrl, | |
| }); | |
| } | |
| else if (configService.get<Log>('LOG').LEVEL.includes('WEBHOOKS')) { | |
| const logData = { | |
| local: `${origin}.sendData-SQS`, | |
| ...message, | |
| }; | |
| this.logger.log(logData); | |
| } | |
Explanation
Flattening if statements nested within else clauses generates code that iseasier to read and expand upon.
…ce `e` with `(e)`"
nolramaf
commented
Sep 1, 2025
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Summary by Sourcery
Introduce a global SQS configuration mode with unified queue provisioning and event routing, add large-payload handling via S3 fallback, and enhance queue management abstractions in SqsController.
New Features:
Enhancements:
Build: