Fix/rabbitmq schema validation graphql introspection#557
Conversation
…/P99 Replace average-based latency tracking with histogram percentile buckets so sparse high-latency outliers remain visible in monitoring. Closes Pi-Defi-world#413
Ensure pooled SMTP connections are closed after batch and single sends, with pool metrics and leak detection to prevent connection exhaustion. Closes Pi-Defi-world#423
Require typ JWT on verify and sign paths so OAuth access tokens (e.g. at+JWT) from the same issuer cannot be accepted as application JWTs. Closes Pi-Defi-world#416
…lover Reconnect and resend failed commands when a promoted replica briefly returns READONLY, preventing cache write failures until failover completes. Closes Pi-Defi-world#412
…express-async-errors
fix: patch Express 4 router with express-async-errors (Pi-Defi-world#420)
…odemailer-transport-pool-leak fix(email): close nodemailer SMTP transport pool after batch sends
…wt-typ-header-validation fix(auth): validate JWT typ header to prevent token confusion
…adonly-failover-retry fix(cache): retry Redis writes on READONLY errors during Sentinel failover
…pi-response-time-histogram feat(metrics): add per-endpoint response time histograms with P50/P95/P99
- Add PG_WAL_BACKUP_CONFIGURED and PG_WAL_BACKUP_PROVIDER env vars to env.ts - connectWithRetry() throws in production when WAL backup is not configured - Log backup provider at startup when configured; warn when absent - Add walBackup guard tests (walBackup.test.ts) - Document vars in .env.example
…-guard fix(Pi-Defi-world#381): add WAL backup configuration guard
…t-length-validation fix(Pi-Defi-world#449): validate Content-Length against actual body size
…icts and missing mocks - Remove duplicate bulkTransfer key in env.ts (TS1117 compile error) - Resolve merge conflict in src/index.ts (remove duplicate AppError import, add execSync import, keep express-async-errors) - Add CORS_ORIGIN to test REQUIRED_ENV (production guard requires it) - Add $extends mock to PrismaClient mock (needed when PRISMA_ACCELERATE_URL set) - Extract shared mockPrismaClient factory to avoid repetition
… logger Set independent console and file transport levels so debug output from stellar-sdk and app code no longer floods production log aggregators. Co-authored-by: Cursor <cursoragent@cursor.com>
Add exchangeRateService with separate positive/negative TTLs so 5xx, timeout, and network failures are not retried on every request. Co-authored-by: Cursor <cursoragent@cursor.com>
Co-authored-by: Cursor <cursoragent@cursor.com>
Add dateUtils with Intl-based timezone handling and wire salary schedules, withdrawal timing, savings lock dates, and limit windows to BUSINESS_TIMEZONE instead of hardcoded UTC or server-local offsets. Co-authored-by: Cursor <cursoragent@cursor.com>
…wallet ops Track walletVersion on users, expose ETag on wallet/balance reads, and require If-Match for wallet mutations and transfers to prevent lost updates. Co-authored-by: Cursor <cursoragent@cursor.com>
…-guard fix(Pi-Defi-world#381): WAL backup guard — fix test failures from merge conflicts and missing mocks
…ransport-levels fix(Pi-Defi-world#398): configure per-transport log levels in Winston logger
…e-negative-cache fix(Pi-Defi-world#404): cache negative exchange rate API responses
…e-utils fix(Pi-Defi-world#408): use IANA timezone-aware business date utilities
…concurrency fix(Pi-Defi-world#407): add If-Match ETag optimistic concurrency for wallet ops
This commit implements schema validation for all RabbitMQ message producers and consumers to prevent malformed messages from accumulating in queues and blocking subsequent processing. Changes: - Add Zod schemas for all message types (escrow, lending pool, savings vault, audit logs, OTP, notifications, webhooks) - Create validation utilities with DLQ support for invalid messages - Implement BaseProducer class with built-in validation - Replace direct queue publishing with validated producers - Update consumers to validate incoming messages before processing - Add message envelope with versioning and unique message IDs - Add dead letter support for validation failures Files Added: - src/types/rabbitmq-schemas.ts - Shared Zod schemas - src/utils/rabbitmq-validation.ts - Validation utilities - src/jobs/producers/ - Producer classes with validation Files Modified: - package.json - Add uuid and update zod - src/jobs/acbu_escrow_event_listener.ts - src/jobs/acbu_lending_pool_event_listener.ts - src/jobs/acbu_savings_vault_event_listener.ts - src/jobs/auditConsumer.ts - src/jobs/notificationConsumer.ts - src/jobs/webhookConsumer.ts Dependencies: - zod: ^3.25.76 - Schema validation - uuid: ^14.0.1 - Message ID generation Benefits: - Prevents malformed messages from entering queues - Invalid messages are sent to DLQ with clear error reasons - Consistent validation across all producers and consumers - Message versioning enables future schema evolution - Improved error logging and observability Closes Pi-Defi-world#399
…world#400) Add defensive middleware to block GraphQL paths and introspection attempts in production. Blocks common GraphQL endpoints, __schema/__type keywords in request bodies and query params, with logging for monitoring. Closes Pi-Defi-world#400
|
Warning Review limit reached
More reviews will be available in 10 minutes and 30 seconds. Learn how PR review limits work. Your organization has used up its prepaid credits, and credit purchases are no longer available. Enable the review add-on in the billing tab to keep reviews running — you're only billed for reviews past your plan's rate limits ($0.25/file). ⌛ How to resolve this issue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based credits. 🚦 How do rate limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please see our Fair Usage Limits Policy for further information. ℹ️ Review info⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Run ID: ⛔ Files ignored due to path filters (1)
📒 Files selected for processing (11)
📝 WalkthroughWalkthroughThis PR adds wallet optimistic concurrency via ETags (new ChangesFull Feature Set
Sequence Diagram(s)sequenceDiagram
participant Client
participant walletController
participant walletStateService
participant Horizon
participant Prisma
Client->>walletController: GET /wallet
walletController->>walletStateService: getWalletState(userId)
walletStateService->>Prisma: findUnique(stellarAddress, walletVersion)
Prisma-->>walletStateService: {stellarAddress, walletVersion: 3}
walletStateService->>Horizon: loadAccount(stellarAddress)
Horizon-->>walletStateService: account balances
walletStateService-->>walletController: {balance, wallet_version: 3}
walletController-->>Client: 200 ETag:"3" + wallet state
Client->>walletController: PUT /wallet/address (If-Match:"3")
walletController->>walletStateService: updateWalletWithConcurrency(userId, "3", data)
walletStateService->>Prisma: updateMany where walletVersion=3, increment
alt version matched
Prisma-->>walletStateService: {count: 1}
walletStateService-->>walletController: newVersion=4
walletController-->>Client: 200 ETag:"4"
else concurrent update
Prisma-->>walletStateService: {count: 0}
walletController-->>Client: 412 PRECONDITION_FAILED
end
Estimated code review effort🎯 5 (Critical) | ⏱️ ~120 minutes Possibly related issues
Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 3 | ❌ 2❌ Failed checks (2 warnings)
✅ Passed checks (3 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
Note
Due to the large number of review comments, Critical severity comments were prioritized as inline comments.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/controllers/investmentController.ts (1)
119-135: 🔒 Security & Privacy | 🔴 Critical | ⚡ Quick winRequire an auth scope before listing withdrawal requests.
Unlike
postInvestmentWithdrawRequest, this path proceeds when bothuserIdandorganizationIdare missing. That makeswherebecome{ organizationId: null }, which can expose user-scoped withdrawal rows instead of rejecting the request.Suggested guard
const userId = req.apiKey?.userId ?? null; const organizationId = req.apiKey?.organizationId ?? null; + if (!userId && !organizationId) { + throw new AppError("User or organization context required", 401); + } const query = getWithdrawRequestsQuerySchema.safeParse(req.query);🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/controllers/investmentController.ts` around lines 119 - 135, The withdrawal listing flow in the investmentController handler currently allows requests with neither userId nor organizationId, causing the where filter to fall back to organizationId: null instead of rejecting the call. Add an explicit auth-scope guard near the existing req.apiKey checks in this handler so it fails fast when both scopes are missing, similar to the protection used in postInvestmentWithdrawRequest. Keep the cursor decoding and query logic in the same path, but only proceed after confirming at least one valid scope is present.
🟠 Major comments (27)
prisma/migrations/20260624120000_add_user_wallet_version/migration.sql-1-1 (1)
1-1: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winFail loudly or enforce the column contract explicitly.
Line 1 can silently preserve schema drift: if
wallet_versionalready exists but is nullable or lacksDEFAULT 0, this migration no-ops whileprisma/schema.prismaassumes a non-nullIntwith a default. That weakens the concurrency contract and can surface as null/version mismatches at runtime.Suggested migration shape
-ALTER TABLE "users" ADD COLUMN IF NOT EXISTS "wallet_version" INTEGER NOT NULL DEFAULT 0; +ALTER TABLE "users" ADD COLUMN "wallet_version" INTEGER NOT NULL DEFAULT 0;If you need idempotence, keep the conditional add but follow it with explicit constraint/default enforcement:
+ALTER TABLE "users" ALTER COLUMN "wallet_version" SET DEFAULT 0; +UPDATE "users" SET "wallet_version" = 0 WHERE "wallet_version" IS NULL; +ALTER TABLE "users" ALTER COLUMN "wallet_version" SET NOT NULL;🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@prisma/migrations/20260624120000_add_user_wallet_version/migration.sql` at line 1, The migration for wallet_version can silently leave an existing column in the wrong shape, so enforce the contract explicitly instead of relying on only adding it conditionally. Update the migration so the users.wallet_version column is guaranteed to be non-null with a default of 0 even when it already exists, or make the migration fail loudly if the existing column does not match the expected contract. Use the migration SQL in the add_user_wallet_version change to align the database with the prisma/schema.prisma expectation for wallet_version.src/utils/walletConcurrency.ts-20-25 (1)
20-25: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winTighten ETag parsing to digits-only.
Lines 20-25 currently accept malformed validators such as
"7junk"becauseparseInt()stops at the first non-digit. For an optimistic-concurrency precondition, that means an invalidIf-Matchvalue can still satisfy version7.Suggested fix
-const ETAG_PATTERN = /^W\/"(.+)"$|^"(.+)"$/; +const ETAG_PATTERN = /^(?:W\/)?"([0-9]+)"$/; ... - const match = raw.match(ETAG_PATTERN); - const token = match?.[1] ?? match?.[2]; + const match = raw.match(ETAG_PATTERN); + const token = match?.[1]; if (!token) return null; - const version = Number.parseInt(token, 10); - return Number.isInteger(version) && version >= 0 ? version : null; + const version = Number(token); + return Number.isSafeInteger(version) ? version : null;🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/utils/walletConcurrency.ts` around lines 20 - 25, The ETag parsing in the token-to-version logic is too permissive because Number.parseInt allows trailing non-digit characters like "7junk". Update the parsing in the walletConcurrency utility to accept only fully numeric validators by verifying the extracted token is digits-only before converting it, and keep the existing non-negative integer check in place. Use the token extraction flow around ETAG_PATTERN, match, and the return logic to locate the fix.src/services/transfer/transferService.ts-110-110 (1)
110-110: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winDo not require
If-Matchfor all transfer callers.
reserveWalletVersionthrows428whenoptions?.ifMatchis missing, butprocessSalaryBatchcallscreateTransferwithout options, so salary transfers will now fail before recipient resolution. Gate this reservation to user-initiated HTTP flows or add a separate internal transfer path that does not require wallet ETags.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/services/transfer/transferService.ts` at line 110, The unconditional reserveWalletVersion call in createTransfer is forcing every transfer path to supply options?.ifMatch, which breaks internal callers like processSalaryBatch before recipient resolution. Update transferService.createTransfer to only reserve the wallet version for user-initiated HTTP flows that actually provide an ETag, or split the logic into a separate internal transfer path that skips the If-Match requirement while preserving the existing behavior for external requests.src/services/wallet/walletStateService.ts-18-21 (1)
18-21: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winBound or sweep the balance cache.
Expired entries remain in
balanceCacheindefinitely for users who do not mutate their wallet again, so a long-lived API process can accumulate one entry per queried wallet/environment. Add a small max size or prune expired entries beforeset.Also applies to: 163-167, 180-184
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/services/wallet/walletStateService.ts` around lines 18 - 21, The balance cache in walletStateService is unbounded and expired entries can linger forever, so add cache hygiene around balanceCache usage. Update the balance write path in the wallet state service (including the helper logic near the referenced cache updates) to prune expired entries before inserting and enforce a small maximum size/eviction policy so long-lived API processes do not accumulate one entry per wallet/environment. Keep the fix localized to the balanceCache Map and the functions that read/write it.src/controllers/walletController.ts-94-100 (1)
94-100: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winUse the existing Stellar address validator here.
This accepts any 56-character string starting with
G, so invalid public keys can be persisted. Reuse the same strict validator used by wallet service before callingupdateWalletWithConcurrency.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/controllers/walletController.ts` around lines 94 - 100, The walletController validation currently uses a loose Zod string check for stellar_address, which can let invalid public keys through. Update the request parsing in walletController to reuse the existing strict Stellar address validator already used by the wallet service before calling updateWalletWithConcurrency, so the same validation logic is applied consistently and invalid addresses are rejected.src/services/wallet/walletStateService.ts-153-159 (1)
153-159: 🗄️ Data Integrity & Integration | 🟠 MajorPreserve the Horizon balance string instead of round-tripping through
numberto avoid precision loss.
Number.parseFloat(acbuBalance.balance)followed byString()strips trailing zeros and can alter the exact formatting provided by the Stellar ledger (e.g., converting "10.0000000" to "10"). Use the validated string from Horizon directly, falling back to"0"only when no matching ACBU balance exists.Suggested fix
- const stellarNum = acbuBalance ? Number.parseFloat(acbuBalance.balance) : 0; - const displayNum = Number.isFinite(stellarNum) ? stellarNum : 0; + const displayBalance = acbuBalance?.balance ?? "0"; const snapshot: WalletBalanceSnapshot = { - balance: String(displayNum), + balance: displayBalance, currency: "ACBU", stellar_address: user.stellarAddress, - balance_stellar: String(displayNum), + balance_stellar: displayBalance, balance_source: "stellar", };🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/services/wallet/walletStateService.ts` around lines 153 - 159, The ACBU balance snapshot in walletStateService is round-tripping the Horizon balance through Number.parseFloat and String, which can change the exact ledger formatting. Update the balance handling in the WalletBalanceSnapshot निर्माण logic to use the validated acbuBalance.balance string directly, and only fall back to "0" when no matching ACBU balance exists; keep the existing stellar_address and balance_stellar fields aligned with that preserved string.src/controllers/walletController.ts-238-248 (1)
238-248: 🔒 Security & Privacy | 🟠 MajorUse a random per-wallet encryption salt and store it with the ciphertext.
The current implementation derives the KDF salt deterministically as
"acbu-wallet-v1:" + userId. This is a security anti-pattern;scryptrequires a unique, random salt for each encryption operation to prevent rainbow table attacks and ensure high entropy.Update the encryption flow to:
- Generate a random salt:
const salt = crypto.randomBytes(32);- Store the salt alongside the ciphertext (include a version byte for future migration).
- Update
decryptUserStellarSecretto extract the random salt from the stored blob before deriving the key.Current vulnerable pattern
const salt = WALLET_ENC_SALT_PREFIX + userId; // Deterministic! const key = crypto.scryptSync(body.passcode, salt, WALLET_ENC_KEYLEN);Implement a versioned envelope structure:
[version][salt][iv][ciphertext][authTag].🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/controllers/walletController.ts` around lines 238 - 248, The wallet encryption flow in walletController must stop using the deterministic WALLET_ENC_SALT_PREFIX + userId salt and instead generate a random per-wallet salt for each encryption. Update the encryption path that builds encryptedStellarSecret to create a versioned envelope in the order [version][salt][iv][ciphertext][authTag], store the salt in the blob, and derive the key from that random salt with scryptSync. Also update decryptUserStellarSecret to read the version byte and extract the salt before key derivation so the decrypted payload matches the new format.src/controllers/investmentController.ts-18-21 (1)
18-21: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winReject non-decimal and infinite withdrawal amounts.
Number(s) > 0accepts values like"Infinity","1e309", and scientific notation. For a persisted financial amount, keep the accepted wire format decimal-only and bounded to the supported precision.Suggested validation tightening
amount_acbu: z .string() .min(1) - .refine((s) => !Number.isNaN(Number(s)) && Number(s) > 0, "must be positive"), + .refine( + (s) => /^\d+(\.\d{1,7})?$/.test(s) && new Decimal(s).gt(0), + "must be a positive decimal with up to 7 decimal places", + ),🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/controllers/investmentController.ts` around lines 18 - 21, The withdrawal amount validation in the investmentController schema currently relies on Number(s) > 0, which allows Infinity and scientific notation. Tighten the refinement on amount_acbu to accept only decimal-formatted strings within the supported precision and reject non-decimal or infinite values. Update the zod validation near the amount_acbu field so the wire format stays bounded and consistent for persisted financial amounts.src/types/rabbitmq-schemas.ts-111-113 (1)
111-113: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winValidate notification payloads per notification type.
The schema currently accepts any payload with a known
type, butprocessNotificationreads required fields likehealth,status,currency,amount,channel, andamountAcbu. Malformed notification messages can still pass validation and fail later or be silently skipped.Proposed direction
-export const NotificationSchema = z.object({ - type: z.enum(['reserve_alert', 'withdrawal_status', 'investment_withdrawal_ready']), -}).passthrough(); +export const NotificationSchema = z.discriminatedUnion("type", [ + z.object({ + type: z.literal("reserve_alert"), + health: z.string().min(1), + overcollateralizationRatio: z.number(), + }), + z.object({ + type: z.literal("withdrawal_status"), + userId: z.string().min(1), + status: z.string().min(1), + currency: z.string().min(1), + amount: z.number(), + channel: z.array(z.enum(["email", "sms"])).optional(), + }), + z.object({ + type: z.literal("investment_withdrawal_ready"), + userId: z.string().min(1).nullable().optional(), + organizationId: z.string().min(1).nullable().optional(), + amountAcbu: z.number(), + }), +]);🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/types/rabbitmq-schemas.ts` around lines 111 - 113, The NotificationSchema currently only validates the type and allows any extra payload, so malformed messages can reach processNotification and fail later. Update NotificationSchema in rabbitmq-schemas.ts to validate a discriminated union by notification type, and add the required fields for each case used by processNotification such as health, status, currency, amount, channel, and amountAcbu. Keep the existing type values, but make the schema strict enough that only payloads matching the expected shape for reserve_alert, withdrawal_status, and investment_withdrawal_ready are accepted.src/utils/rabbitmq-validation.ts-96-100 (1)
96-100: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winTreat malformed JSON and envelope/queue mismatches as validation failures.
JSON.parseerrors currently bypassMessageValidationError, so some consumers retry poison messages instead of dead-lettering them as invalid. Also,envelope.typeis ignored, allowing a valid envelope for one queue to be processed from another queue when payload schemas overlap.Proposed fix
- const raw = JSON.parse(content.toString()); + let raw: unknown; + try { + raw = JSON.parse(content.toString()); + } catch { + throw new MessageValidationError(queue, [ + { code: z.ZodIssueCode.custom, path: [], message: 'Invalid JSON' }, + ]); + } + const envelope = MessageEnvelopeSchema.parse(raw); + + if (envelope.type !== queue) { + throw new MessageValidationError(queue, [ + { + code: z.ZodIssueCode.custom, + path: ['type'], + message: `Envelope type ${envelope.type} does not match queue ${queue}`, + }, + ]); + } // Validate payload against queue schema🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/utils/rabbitmq-validation.ts` around lines 96 - 100, Malformed JSON and envelope/queue mismatches are not being converted into MessageValidationError, so update the validation flow in the message parsing/validation helper to catch JSON.parse failures and wrap them as validation failures, and also verify envelope.type matches the expected queue before calling validateMessage. Use the existing MessageEnvelopeSchema parsing path and the queue validation logic in rabbitmq-validation.ts so both invalid JSON and wrong-queue envelopes are rejected consistently.src/jobs/auditConsumer.ts-37-38 (1)
37-38: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winPreserve explicit
nullaudit values.
AuditLogSchemaallowsoldValue/newValueto benull, but?? undefineddrops explicit nulls and loses audit detail for fields intentionally cleared to null.Proposed fix
- oldValue: validatedEntry.oldValue ?? (undefined as any), - newValue: validatedEntry.newValue ?? (undefined as any), + oldValue: + validatedEntry.oldValue === undefined + ? undefined + : (validatedEntry.oldValue as any), + newValue: + validatedEntry.newValue === undefined + ? undefined + : (validatedEntry.newValue as any),🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/jobs/auditConsumer.ts` around lines 37 - 38, The audit mapping in auditConsumer’s entry validation is converting explicit null values into undefined because of the nullish coalescing fallback. Update the handling of validatedEntry.oldValue and validatedEntry.newValue so AuditLogSchema nulls are preserved exactly as provided, while still allowing undefined when the field is truly absent; use the existing audit normalization path in auditConsumer to keep the distinction intact.src/utils/rabbitmq-validation.ts-33-36 (1)
33-36: 🔒 Security & Privacy | 🟠 Major | ⚡ Quick winDo not log message payloads on validation failure.
This path can log sensitive fields from invalid OTP, audit, notification, or webhook messages. Log schema errors and safe metadata instead.
Proposed fix
logger.error('Message validation failed', { queue, errors: error.errors, - payload: JSON.stringify(payload).substring(0, 500), + payloadKeys: + payload && typeof payload === 'object' + ? Object.keys(payload as Record<string, unknown>) + : undefined, });🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/utils/rabbitmq-validation.ts` around lines 33 - 36, The validation failure log in rabbitmq-validation should not include the raw message payload, since it can expose sensitive data. Update the logging in the message validation path to remove the payload field entirely and keep only schema error details plus safe metadata such as queue name and any non-sensitive identifiers. Use the existing validation/error handling in the rabbitmq-validation utility to ensure Message validation failed still provides enough context without serializing payload contents.src/jobs/acbu_savings_vault_event_listener.ts-37-48 (1)
37-48: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winDon't drop savings-vault events when publish fails.
Lines 37-48 log
savingsVaultEventProducer.publish()failures and then continue. That silently drops the contract event on validation or broker errors, bypassing the DLQ/reliability goal of this change. Please dead-letter the raw event or bubble the error into the listener retry flow.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/jobs/acbu_savings_vault_event_listener.ts` around lines 37 - 48, The failure path in acbu_savings_vault_event_listener’s publish flow is swallowing errors after savingsVaultEventProducer.publish(), which drops events instead of triggering reliability handling. Update the catch in the listener logic to either dead-letter the raw event explicitly or rethrow/bubble the error so the listener retry flow can take over, and keep the existing validated publish path and logging around the relevant event and producer symbols.src/jobs/acbu_escrow_event_listener.ts-36-47 (1)
36-47: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winDon't drop escrow events when publish fails.
Lines 36-47 catch
escrowEventProducer.publish()failures and then return. That means validation/broker errors discard the source event with no retry or DLQ record, which leaves downstream escrow processing out of sync. Please either dead-letter the raw event here or propagate the failure into the listener's retry path.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/jobs/acbu_escrow_event_listener.ts` around lines 36 - 47, The `escrowEventProducer.publish()` failure path in `acbu_escrow_event_listener` is swallowing the error after logging, which drops the source escrow event; update the `catch` block so failures are either sent to a dead-letter flow with the raw event payload or rethrown into the listener retry path. Use the existing `logger.error` and `validatedEvent`/`event` context to preserve the payload and ensure `publish` failures are not silently discarded.src/jobs/acbu_lending_pool_event_listener.ts-37-48 (1)
37-48: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winDon't drop lending-pool events when publish fails.
Lines 37-48 swallow
lendingPoolEventProducer.publish()failures after logging them. Any schema error or transient RabbitMQ failure now loses the originating contract event without a DLQ entry or retry signal. This path should dead-letter the raw event or rethrow so the listener can retry.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/jobs/acbu_lending_pool_event_listener.ts` around lines 37 - 48, The publish failure in lendingPoolEventProducer.publish() is being swallowed after logging, which causes raw lending-pool events to be lost. Update acbu_lending_pool_event_listener.ts in the listener path around the publish/try-catch so that failures either rethrow to let the consumer retry or explicitly dead-letter the raw event before returning. Use the existing logger and the lendingPoolEventProducer.publish flow to keep the event from disappearing on schema or RabbitMQ errors.src/jobs/producers/BaseProducer.ts-18-27 (1)
18-27: 🔒 Security & Privacy | 🟠 Major | ⚡ Quick winStop logging full message bodies from the shared producer path.
Lines 18-27 write the entire payload on both success and failure. Because this is the base class for queue producers, that leaks message contents into application logs and makes validation failures a privacy/compliance problem.
publishValidatedMessage()already logs queue/message metadata, so this should stick to identifiers rather than bodies.Suggested change
logger.debug("Message published successfully", { queue: this.queue, - payload: validatedPayload, }); } catch (error) { logger.error("Failed to publish message", { queue: this.queue, error: error instanceof Error ? error.message : String(error), - payload, }); throw error; }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/jobs/producers/BaseProducer.ts` around lines 18 - 27, The shared producer path in publishValidatedMessage() is logging full message bodies on both success and failure, which leaks payload contents. Update BaseProducer so the logger.debug and logger.error calls keep only queue/message metadata (for example identifiers, counts, or validation status) and remove validatedPayload/payload from the logged context. Use publishValidatedMessage() and this.queue as the main anchors when adjusting the logging shape.src/jobs/producers/BaseProducer.ts-11-16 (1)
11-16: 🚀 Performance & Scalability | 🟠 Major | ⚡ Quick winCache queue declaration instead of asserting on every publish.
Lines 12-16 currently turn each publish into a topology setup call.
assertQueueWithDLQ()performs multiple broker RPCs, so every message pays for re-asserting an already-existing queue/DLQ pair. On these event streams, that will become a throughput bottleneck fast. Assert once per producer (or at startup) and reuse the promise.Suggested change
export abstract class BaseProducer<T> { protected abstract queue: string; protected abstract validate(payload: T): T; + private queueReady?: Promise<void>; + + protected async ensureQueue(): Promise<void> { + if (!this.queueReady) { + this.queueReady = assertQueueWithDLQ(this.queue) + .then(() => undefined) + .catch((error) => { + this.queueReady = undefined; + throw error; + }); + } + + return this.queueReady; + } async publish(payload: T, options?: { persistent?: boolean; priority?: number }): Promise<void> { try { - // Ensure queue exists with DLQ - await assertQueueWithDLQ(this.queue); + await this.ensureQueue(); // Validate and publish const validatedPayload = this.validate(payload);🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/jobs/producers/BaseProducer.ts` around lines 11 - 16, Cache the queue declaration in BaseProducer so publish does not re-run topology setup on every message. Update the publish path around validate/publish to reuse a one-time assertQueueWithDLQ() promise (or move the assertion to producer initialization/startup) and have the producer methods reference that cached setup before calling publishValidatedMessage().src/services/notification/notificationService.ts-96-114 (1)
96-114: 🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
sendEmailBatch()exposes an all-or-nothing API over partial side effects.Lines 101-114 reject the whole batch on the first provider error even though earlier recipients may already have been sent. Since
src/jobs/notificationConsumer.ts:99uses this from a queue consumer, a retry can resend the already-delivered subset.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/services/notification/notificationService.ts` around lines 96 - 114, The sendEmailBatch() flow is all-or-nothing even though some messages may already be sent before an error, which can cause duplicate deliveries on retry from notificationConsumer. Update sendEmailBatch() to handle per-message failures without throwing away the whole batch: keep the existing SMTP path behavior in sendSmtpEmailBatch(), but in the non-SMTP loop catch and log failures per sendEmail() call while continuing with the remaining messages, and only surface or aggregate failures in a way that preserves which recipients succeeded. Use the sendEmailBatch() and sendEmail() symbols to locate the batch and per-recipient paths.src/services/cache/redisService.ts-149-156 (1)
149-156: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winDon't silently drop expirations when
ttlSecondsis0.Line 155 uses a truthy check, so
set(key, value, 0)stores the key without any TTL instead of expiring or rejecting it. For cache and rate-limit keys, that turns a short-lived entry into a permanent one.Suggested fix
async set( key: string, value: string, ttlSeconds?: number, ): Promise<"OK" | null> { return this.executeWithReadonlyRetry(async () => { - if (ttlSeconds) { + if (ttlSeconds !== undefined) { + if (!Number.isInteger(ttlSeconds) || ttlSeconds <= 0) { + throw new Error("ttlSeconds must be a positive integer"); + } return this.getClient().set(key, value, "EX", ttlSeconds); } return this.getClient().set(key, value); }); }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/services/cache/redisService.ts` around lines 149 - 156, The TTL handling in `RedisService.set` is using a truthy check, so `ttlSeconds = 0` skips expiration and stores the key permanently. Update the `set` method to treat `0` explicitly instead of relying on truthiness, and either apply the expiration or reject invalid TTL values consistently. Use the `set` method and `executeWithReadonlyRetry` path to locate the fix, and ensure cache/rate-limit callers do not lose expiration semantics when passing zero.src/services/email/emailService.ts-119-123 (1)
119-123: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winTransport cleanup can mask the real delivery outcome.
Lines 120-123 let
closeSmtpTransport()throw out of thefinally. If the emails were already accepted, the caller still sees a failure and may retry the batch; if a send already failed, the close error hides the real cause. It also skips the metric cleanup on that path.Suggested fix
export async function sendSmtpEmailBatch( messages: SmtpEmailMessage[], ): Promise<void> { if (messages.length === 0) { return; } poolMetrics.batchesInFlight += 1; const transport = createSmtpTransport(); + let sendError: unknown; try { for (const message of messages) { await transport.sendMail({ from: config.notification.emailFrom, to: message.to, subject: message.subject, text: message.body, }); } + } catch (error) { + sendError = error; + throw error; } finally { - await closeSmtpTransport(transport, "batch"); - poolMetrics.batchesInFlight -= 1; - poolMetrics.totalBatchesCompleted += 1; - recordPoolLeakIfDetected("batch"); + try { + await closeSmtpTransport(transport, "batch"); + } catch (closeError) { + if (!sendError) { + throw closeError; + } + logger.error("SMTP transport cleanup failed after send failure", { + closeError, + }); + } finally { + poolMetrics.batchesInFlight = Math.max(0, poolMetrics.batchesInFlight - 1); + if (!sendError) { + poolMetrics.totalBatchesCompleted += 1; + } + recordPoolLeakIfDetected("batch"); + } } }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/services/email/emailService.ts` around lines 119 - 123, The batch cleanup in emailService’s finally block lets closeSmtpTransport() throw and overwrite the real send result, which can cause false batch failures and skip metric cleanup. Update the batch flow around closeSmtpTransport, poolMetrics.batchesInFlight, poolMetrics.totalBatchesCompleted, and recordPoolLeakIfDetected so transport shutdown is best-effort and never masks the original delivery outcome. Catch and log cleanup errors separately, and ensure the metric adjustments still run regardless of whether SMTP closing succeeds.src/services/ai/openaiGuard.ts-212-249 (1)
212-249: 🗄️ Data Integrity & Integration | 🟠 MajorTimed-out attempts still run to completion, causing duplicate billing.
Promise.race()stops awaiting the result but does not cancel the in-flight HTTP request. If a timeout occurs, the original OpenAI request continues processing in the background. A subsequent retry sends a new request, potentially charging for two completions while only one is used.Update the SDK call to accept an
AbortControllersignal passed as thesignaloption, aborting it on timeout or successful resolution of the race.const controller = new AbortController(); const apiPromise = client.chat.completions.create({ model, messages, max_tokens: maxTokens, signal: controller.signal, // <-- Pass signal }); const timeoutPromise = new Promise((_, rej) => setTimeout(() => { controller.abort(); // <-- Abort on timeout rej(new Error("OpenAI request timed out")); }, timeoutMs), );Include logic to abort the controller on success after the race resolves to prevent lingering requests.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/services/ai/openaiGuard.ts` around lines 212 - 249, The retry logic in callWithRetries() leaves timed-out OpenAI requests running, which can duplicate completions and billing. Update the client.chat.completions.create call to use an AbortController signal, and abort that controller when the timeoutPromise wins the Promise.race in openaiGuard.ts. Also ensure the controller is aborted after a successful race resolution so no in-flight request lingers, and keep the existing retry/error handling around isRetryableError, lastErr, and logger.warn intact.src/services/rates/exchangeRateService.ts-9-16 (1)
9-16: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winGuard TTL parsing against invalid values.
If either env var is empty or non-numeric,
parseInt(...)yieldsNaN,expiresAtbecomesNaN, andgetCachedExchangeRate()never evicts that entry becauseDate.now() >= NaNis always false. A bad negative TTL can pin a cachednullindefinitely. Prefer validated config values here, or clamp to a finite positive fallback before writing the cache entry.Suggested fix
-const POSITIVE_TTL_MS = parseInt( - process.env.EXCHANGE_RATE_CACHE_TTL_MS || "60000", - 10, -); -const NEGATIVE_TTL_MS = parseInt( - process.env.EXCHANGE_RATE_NEGATIVE_CACHE_TTL_MS || "30000", - 10, -); +function parsePositiveTtl(raw: string | undefined, fallbackMs: number): number { + const parsed = Number(raw); + return Number.isFinite(parsed) && parsed > 0 ? parsed : fallbackMs; +} + +const POSITIVE_TTL_MS = parsePositiveTtl( + process.env.EXCHANGE_RATE_CACHE_TTL_MS, + 60_000, +); +const NEGATIVE_TTL_MS = parsePositiveTtl( + process.env.EXCHANGE_RATE_NEGATIVE_CACHE_TTL_MS, + 30_000, +);🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/services/rates/exchangeRateService.ts` around lines 9 - 16, The TTL constants in exchangeRateService are parsed directly from environment variables, so empty or non-numeric values can produce NaN and break cache eviction in getCachedExchangeRate(). Update POSITIVE_TTL_MS and NEGATIVE_TTL_MS to use validated finite numeric values with safe fallbacks before they are used to compute expiresAt, and ensure any invalid config is clamped to a positive default so cached entries cannot become permanent.src/services/rates/exchangeRateService.ts-37-46 (1)
37-46: 🩺 Stability & Availability | 🟠 MajorDon't negative-cache non-Axios exceptions.
if (!axios.isAxiosError(error)) return true;treats local code defects (e.g.,TypeError, missing imports) as transient upstream failures. This causesfetchFromExternalApi()bugs to be negative-cached (null), forcing callers to silently degrade for the negative cache TTL instead of surfacing the error on subsequent requests.Non-Axios errors should not populate the negative cache; they indicate application bugs that need fixing, not transient conditions to cache.
Suggested fix
export function isRetryableRateApiError(error: unknown): boolean { - if (!axios.isAxiosError(error)) return true; + if (!axios.isAxiosError(error)) return false; const status = error.response?.status; if (status !== undefined && status >= 500) return true; if (status === 408 || status === 429) return true; if (error.code === "ECONNABORTED" || error.code === "ETIMEDOUT") return true; if (!error.response) return true; return false; }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/services/rates/exchangeRateService.ts` around lines 37 - 46, The retryable error check in isRetryableRateApiError is treating non-Axios exceptions as transient, which causes fetchFromExternalApi to negative-cache application bugs. Update the handling so only Axios-based failures with retryable conditions return true, and make non-Axios errors return false (or otherwise bypass negative caching) so unexpected local defects are surfaced instead of cached as null.src/config/env.ts-116-148 (1)
116-148: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winValidate the new Redis/SMTP envs in
envSchemainstead of parsing them ad hoc.These settings currently bypass
envSchema, so typos become late runtime failures (port: NaN, malformed sentinel entries,NOTIFICATION_EMAIL_PROVIDER="smpt", etc.) instead of failing fast during boot.Suggested direction
+ REDIS_MAX_RETRIES_PER_REQUEST: z.coerce.number().int().min(0).default(3), + REDIS_READONLY_RETRY_ATTEMPTS: z.coerce.number().int().min(0).default(3), + REDIS_READONLY_RETRY_DELAY_MS: z.coerce.number().int().min(0).default(100), + NOTIFICATION_EMAIL_PROVIDER: z.enum(["sendgrid", "ses", "smtp", "log"]).default("log"), + SMTP_PORT: z.coerce.number().int().min(1).max(65535).default(587), + SMTP_MAX_CONNECTIONS: z.coerce.number().int().min(1).default(5), + SMTP_MAX_MESSAGES: z.coerce.number().int().min(1).default(100),Then derive
config.redis/config.notification.smtpfromenv, notprocess.env.Also applies to: 334-348
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/config/env.ts` around lines 116 - 148, The Redis and SMTP settings are being parsed directly from process.env instead of being validated by envSchema, so malformed values slip through until runtime. Move the new Redis/SMTP fields into envSchema validation, then build config.redis and config.notification.smtp from the validated env object rather than re-reading process.env. Use the existing env.ts config setup and the Redis parsing block as the place to update, so bad values like malformed sentinels or typoed provider names fail fast during boot.src/config/mongodb.ts-30-35 (1)
30-35: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winAdd jitter to the MongoDB retry delay.
BASE_DELAY_MS * 2 ** (attempt - 1)makes every instance reconnect on the same schedule after a shared outage, so this loop can still herd traffic back onto MongoDB all at once.Suggested fix
- const delay = BASE_DELAY_MS * 2 ** (attempt - 1); + const cappedBackoff = BASE_DELAY_MS * 2 ** (attempt - 1); + const delay = Math.floor(Math.random() * cappedBackoff);🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/config/mongodb.ts` around lines 30 - 35, The retry backoff in the MongoDB connection loop is deterministic, which can cause many instances to reconnect at the same time after an outage. Update the delay calculation in the MongoDB retry logic to add random jitter on top of BASE_DELAY_MS * 2 ** (attempt - 1), while keeping the existing logger.warn, sanitizeMongoUri, and retry flow intact so reconnects are spread out across instances.src/controllers/mintController.test.ts-128-148 (1)
128-148: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winAssert the flattened validation payload, not
AppError.message.For
"JPY",depositFromBasketCurrencyfailsdepositBodySchema.safeParse(...)and wraps that asnew AppError("Invalid request", ...)insrc/controllers/mintController.tsLines 289-296. The expectation on Line 147 will therefore fail even when the validation is working. Check the attached field error forcurrency, or assert"Invalid request"here.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/controllers/mintController.test.ts` around lines 128 - 148, The test for depositFromBasketCurrency is asserting the wrong error text: the schema validation failure is wrapped by the controller into AppError("Invalid request", ...) rather than exposing the raw field message. Update the expectation in mintController.test to assert the wrapped message or inspect the flattened validation payload on the AppError details for the currency field, using depositFromBasketCurrency and AppError as the key symbols to locate the behavior.src/middleware/metrics.ts-114-132 (1)
114-132: 🩺 Stability & Availability | 🟠 MajorPrevent unbounded metric cardinality from unmatched paths.
The fallback logic in
src/middleware/metrics.tsLine 120 usesreq.pathdirectly whenreq.routeis missing (e.g., 404s). This allows any unique path (like/users/123,/users/xyz,/attack/long/path) to create a new metric entry inhistogramsByEndpointand thehttp.routeOTel attribute. Since this middleware is globally registered, this causes unbounded memory growth and metric cardinality explosion.Replace the fallback with a fixed label (e.g.,
UNKNOWNor/*) to collapse unmatched requests into a single metric series.Current vulnerable logic
if (req.route?.path) { const base = req.baseUrl || ""; return `${req.method} ${base}${req.route.path}`; } return `${req.method} ${req.baseUrl}${req.path}`; // ⚠️ Raw path creates new metrics for every 404🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/middleware/metrics.ts` around lines 114 - 132, The fallback in normalizeEndpoint currently uses req.path for unmatched routes, which creates unbounded metric cardinality for 404s and other misses. Update normalizeEndpoint in src/middleware/metrics.ts to return a fixed label for missing req.route?.path (for example UNKNOWN or /*) instead of the raw path, so histogramsByEndpoint and the http.route attribute in recordResponseTime collapse all unmatched requests into one series. Keep the existing behavior for matched routes using req.route.path and req.baseUrl.
🧹 Nitpick comments (2)
src/types/prisma-extensions.d.ts (1)
3-6: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAvoid erasing the augmented Prisma surface with
any.Line 5 turns every
prisma.bulkTransferJob.*call into uncheckedany, which defeats the point of adding the augmentation in the first place. Please bind this property to the generated Prisma delegate/extension type so bad query shapes still fail at compile time.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/types/prisma-extensions.d.ts` around lines 3 - 6, The PrismaClient augmentation for bulkTransferJob is currently typed as any, which removes compile-time safety for prisma.bulkTransferJob usage. Update the Prisma module augmentation in prisma-extensions.d.ts so bulkTransferJob uses the generated delegate/extension type from `@prisma/client` instead of any, and keep the declaration aligned with the PrismaClient surface so invalid query shapes still fail type-checking.tests/asyncErrors.test.ts (1)
5-8: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winTest the async-error behavior instead of grepping the source file.
This only proves
src/index.tscurrently contains that exact string. It still passes if the patch is never exercised in the real bootstrap path, and it fails on harmless refactors like moving the import to another bootstrap module. A tiny Express app with an async route that throws would validate the behavior this test actually cares about.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/asyncErrors.test.ts` around lines 5 - 8, Replace the source-file grep in asyncErrors.test.ts with a behavioral test that exercises the async error handling path end-to-end. Update the test to spin up a minimal Express app that uses the bootstrap path from src/index.ts (or the same async-error middleware setup) and verify an async route throw is forwarded to the error handler instead of checking for the literal express-async-errors import string.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 1fd0fae8-76cb-4851-b200-9e56cbf8f3af
⛔ Files ignored due to path filters (1)
pnpm-lock.yamlis excluded by!**/pnpm-lock.yaml
📒 Files selected for processing (80)
.env.example.github/pull_request_template.mdREADME.mdpackage.jsonprisma/migrations/20260624120000_add_user_wallet_version/migration.sqlprisma/schema.prismasrc/config/database.tssrc/config/env.tssrc/config/investment.tssrc/config/logger.tssrc/config/mongodb.tssrc/config/savings.tssrc/controllers/investmentController.tssrc/controllers/mintController.test.tssrc/controllers/mintController.tssrc/controllers/savingsController.tssrc/controllers/schemas.tssrc/controllers/transactionController.tssrc/controllers/transferController.tssrc/controllers/userController.tssrc/controllers/walletController.tssrc/index.tssrc/jobs/acbu_escrow_event_listener.tssrc/jobs/acbu_lending_pool_event_listener.tssrc/jobs/acbu_savings_vault_event_listener.tssrc/jobs/auditConsumer.tssrc/jobs/notificationConsumer.tssrc/jobs/producers/AuditLogProducer.tssrc/jobs/producers/BaseProducer.tssrc/jobs/producers/EscrowEventProducer.tssrc/jobs/producers/LendingPoolEventProducer.tssrc/jobs/producers/SavingsVaultEventProducer.tssrc/jobs/producers/index.tssrc/jobs/webhookConsumer.tssrc/middleware/auth.test.tssrc/middleware/auth.tssrc/middleware/authMiddleware.test.tssrc/middleware/authMiddleware.tssrc/middleware/bodyParser.tssrc/middleware/metrics.test.tssrc/middleware/metrics.tssrc/middleware/pagination.test.tssrc/middleware/pagination.tssrc/routes/userRoutes.tssrc/services/ai/openaiGuard.tssrc/services/cache/index.tssrc/services/cache/redisService.test.tssrc/services/cache/redisService.tssrc/services/email/emailService.test.tssrc/services/email/emailService.tssrc/services/email/index.tssrc/services/investment/withdrawalTimingService.tssrc/services/limits/limitsService.tssrc/services/notification/index.tssrc/services/notification/notificationService.tssrc/services/oracle/forexClient.tssrc/services/rates/currencyConverter.tssrc/services/rates/exchangeRateService.test.tssrc/services/rates/exchangeRateService.tssrc/services/rates/index.tssrc/services/salary/salaryService.tssrc/services/transfer/transferService.tssrc/services/transfer/types.tssrc/services/wallet/walletService.tssrc/services/wallet/walletStateService.test.tssrc/services/wallet/walletStateService.tssrc/types/prisma-extensions.d.tssrc/types/rabbitmq-schemas.tssrc/types/shims-papaparse.d.tssrc/utils/dateUtils.test.tssrc/utils/dateUtils.tssrc/utils/jwt.tssrc/utils/rabbitmq-validation.tssrc/utils/walletConcurrency.test.tssrc/utils/walletConcurrency.tstests/asyncErrors.test.tstests/jwt.clockskew.test.tstests/logger.transportLevels.test.tstests/walBackup.test.tstsconfig.build.json
💤 Files with no reviewable changes (1)
- src/controllers/userController.ts
| // For rejected transfers, the item status write inside the map never ran, | ||
| // so collect those corrective writes and commit them atomically below (#394). | ||
| const failedItemWrites: Prisma.PrismaPromise<unknown>[] = []; | ||
| results.forEach((r, idx) => { | ||
| if (r.status === "fulfilled" && r.value === "completed") { | ||
| successCount++; | ||
| } else { | ||
| // fulfilled-but-failed or rejected | ||
| failCount++; | ||
| if (r.status === "rejected") { | ||
| logger.error("Salary item transfer failed", { batchId, error: r.reason }); | ||
| // Mark the corresponding item failed; find by position in chunk | ||
| const idx = results.indexOf(r); | ||
| // Index aligns: results[idx] corresponds to chunk[idx]. | ||
| const item = chunk[idx]; | ||
| if (item) { | ||
| await prisma.salaryItem.update({ | ||
| where: { id: item.id }, | ||
| data: { | ||
| status: "failed", | ||
| errorMessage: | ||
| r.reason instanceof Error ? r.reason.message : "Unknown error", | ||
| }, | ||
| }); | ||
| failedItemWrites.push( | ||
| prisma.salaryItem.update({ | ||
| where: { id: item.id }, | ||
| data: { | ||
| status: "failed", | ||
| errorMessage: r.reason instanceof Error ? r.reason.message : "Unknown error", | ||
| }, | ||
| }), | ||
| ); | ||
| } | ||
| } | ||
| } | ||
| }); | ||
|
|
||
| if (failedItemWrites.length > 0) { | ||
| await prisma.$transaction(failedItemWrites); |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🔴 Critical | 🏗️ Heavy lift
Don't collapse status-write failures into transfer failures.
Line 170 treats every rejected settled result as a failed payout and rewrites the item to status: "failed". But the promise can also reject after createTransfer() succeeded, if the salaryItem.update() inside the mapper fails. In that case the money may already be sent, and this fallback records it as failed, so a later retry can pay the same recipient twice.
Suggested direction
- const results = await Promise.allSettled(
- chunk.map(async (item) => {
- const result = await createTransfer({
- senderUserId: batch.userId,
- to: item.recipientAddress,
- amountAcbu: item.amount.toString(),
- });
- await prisma.salaryItem.update({
- where: { id: item.id },
- data: {
- status: result.status,
- transactionId: result.transactionId,
- errorMessage:
- result.status === "failed" ? "Transfer payment failed" : null,
- },
- });
- return result.status;
- }),
- );
+ const results = await Promise.allSettled(
+ chunk.map(async (item) => {
+ try {
+ const transfer = await createTransfer({
+ senderUserId: batch.userId,
+ to: item.recipientAddress,
+ amountAcbu: item.amount.toString(),
+ });
+ return { kind: "transfer_result" as const, item, transfer };
+ } catch (error) {
+ return { kind: "transfer_error" as const, item, error };
+ }
+ }),
+ );
+
+ // Only mark failed when the transfer itself failed.
+ // If persisting the transfer result fails, surface that separately instead of
+ // converting a completed payout into a business-level failure.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/services/salary/salaryService.ts` around lines 161 - 190, The fallback in
salary transfer handling is conflating transfer failures with status-write
failures in salaryService’s result loop, which can incorrectly mark
already-completed payouts as failed. Update the logic around the
`results.forEach` reconciliation so only actual `createTransfer()` failures
produce a `"failed"` item write, and treat rejections from the
`salaryItem.update()` path as a separate persistence error path instead of
rewriting payout status. Use the `results`, `failedItemWrites`, and
`salaryItem.update` flow to distinguish these cases before calling
`prisma.$transaction`.
| @@ -0,0 +1,151 @@ | |||
| import { z } from 'zod'; | |||
There was a problem hiding this comment.
🎯 Functional Correctness | 🔴 Critical | ⚡ Quick win
Import QUEUES before using it in QUEUE_SCHEMAS.
QUEUE_SCHEMAS references QUEUES, but this file only imports z, so the new file will not compile.
Proposed fix
import { z } from 'zod';
+import { QUEUES } from '../config/rabbitmq';Also applies to: 139-149
🧰 Tools
🪛 ESLint
[error] 1-1: Replace 'zod' with "zod"
(prettier/prettier)
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/types/rabbitmq-schemas.ts` at line 1, QUEUE_SCHEMAS references QUEUES but
the symbol is not imported in this module, so the new schema file will not
compile. Update the imports in rabbitmq-schemas.ts to bring in QUEUES from the
module that defines it, alongside z, and ensure QUEUES is available before
QUEUE_SCHEMAS is declared.
…ql-introspection Resolve merge conflicts and integrate upstream changes: - Package updates: Add prom-client, winston-daily-rotate-file, uuid; update zod to v3.25.76 - Environment config: Add WAL backup configuration (Pi-Defi-world#381) and memory leak detection (Pi-Defi-world#436) - Logging: Replace static file transports with DailyRotateFile with retention policies - Middleware: Add correlationMiddleware, userAgentFilter, and request metrics - Security: Add GraphQL introspection blocking and content-encoding validation - Database: Add replica URL fallback and connection pool retry improvements - Audit consumer: Integrate queueConfig for configurable retries with schema validation Conflicts resolved: - package.json (dependencies and pnpm overrides) - src/config/env.ts (WAL backup + memory detection) - src/config/logger.ts (DailyRotateFile integration) - src/controllers/mintController.ts (idempotency and duplicate checks) - src/index.ts (middleware ordering and GraphQL blocking) - src/jobs/auditConsumer.ts (validation + queue config) - src/config/database.ts (replica URL fallback)
|
@Oluwaseyi89 Great news! 🎉 Based on an automated assessment of this PR, the linked Wave issue(s) no longer count against your application limits. You can now already apply to more issues while waiting for a review of this PR. Keep up the great work! 🚀 |
|
@Junman140 PR is ready |
8 participants
Add schema validation for RabbitMQ messages and block GraphQL introspection
Overview
This PR addresses two security and reliability concerns:
Changes
RabbitMQ Schema Validation
GraphQL Introspection Blocking
Files Added
Files Modified
Dependencies Added
Closes #399
Closes #400
Summary by CodeRabbit