Repository navigation
event based user create or role change - #1365
Conversation
|
Important Review skippedAuto reviews are disabled on base/target branches other than the default branch. Please check the settings in the CodeRabbit UI or the You can disable this status message by setting the WalkthroughReplaces per-topic Kafka consumers with a unified startConsumer(kafkaClient) subscribed to EVENTS_TOPIC and CLEAR_INTERNAL_CACHE, centralizes message parsing and error handling, routes EVENTS_TOPIC by eventType to appropriate handlers, and threads tenantCode into user fetch/update flows while normalizing string IDs. Changes
Sequence Diagram(s)sequenceDiagram
autonumber
participant Producer as External Producer
participant Kafka as Kafka Broker
participant Consumer as startConsumer
participant Users as services/users
participant Admin as services/admin
participant OrgAdmin as services/org-admin
participant Utils as utils
Producer->>Kafka: Publish to EVENTS_TOPIC or CLEAR_INTERNAL_CACHE
Kafka-->>Consumer: Deliver message
rect rgba(230,245,255,0.6)
Consumer->>Consumer: guard empty, ensure string, parse JSON
alt topic == EVENTS_TOPIC
Consumer->>Consumer: inspect eventType
alt create / bulk-create
Consumer->>Users: handle create(s) (includes tenantCode)
else delete
Consumer->>Admin: handle delete
else update / bulk-update
Consumer->>Users: handle update(s) (includes tenantCode)
else roleChange
Consumer->>OrgAdmin: handle role change
end
else topic == CLEAR_INTERNAL_CACHE
Consumer->>Utils: internalDel(streamingData.value)
end
Consumer->>Consumer: commit/ack and centralized error logging on exceptions
end
Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 minutes Possibly related PRs
Poem
Pre-merge checks and finishing touches❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
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 full review |
✅ Actions performedFull review triggered. |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (1)
src/envVariables.js (1)
576-585: Clarify messages and (optionally) align naming with existing Kafka env vars
- Messages say “Required …” while
optional: true. Either make them required or adjust the copy to avoid confusion.- Optional: consider prefixing with KAFKA_ for consistency with the rest of the file (e.g., KAFKA_HEALTH_CHECK_TOPIC, KAFKA_MENTORING_TOPIC).
Apply this diff to clarify the messages:
- EVENTS_TOPIC: { - message: 'Required event topic for handling events', + EVENTS_TOPIC: { + message: 'Event topic for handling events', optional: true, default: 'mentoring.events', }, - EVENTS_GROUP_ID: { - message: 'Required event group id', + EVENTS_GROUP_ID: { + message: 'Event consumer group id', optional: true, default: 'mentoring.event', },Please also ensure these new vars are documented in deployment manifests and CI/CD secrets so they don’t get missed in non-default environments.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (2)
src/configs/kafka.js(4 hunks)src/envVariables.js(1 hunks)
🔇 Additional comments (4)
src/configs/kafka.js (4)
12-15: Confirm service API availability and usage in consumer contextVerify that
@services/users.add,@services/admin.userDelete, and@services/org-admin.roleChangeexist and are safe to call from a long‑running consumer (timeouts, idempotency).
25-25: Dedicated consumer group for events looks goodUsing a distinct
EVENTS_GROUP_IDisolates processing. Make sure it’s unique per service deployment to avoid unintended rebalancing.
29-29: Connecting the new event consumer is correctNo issues here.
42-43: Incorrect — keepsubscribe({ topics: [...] })
KafkaJS (modern versions) supports subscribe({ topics: [...] }); the existing call is valid — no change required.Likely an incorrect or invalid review comment.
|
@coderabbitai full review |
✅ Actions performedFull review triggered. |
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Nitpick comments (1)
src/configs/kafka.js (1)
133-133: Remove redundant error logging.The
console.log('----------', error)line adds unnecessary noise to the logs and duplicates the structured logging that follows.- console.log('----------', error)
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (2)
src/configs/kafka.js(4 hunks)src/envVariables.js(1 hunks)
🔇 Additional comments (8)
src/envVariables.js (1)
576-585: LGTM! Well-structured environment variable declarations.The new environment variables for event processing are correctly structured and consistent with the existing pattern. The default values are appropriate and the messages clearly describe their purpose.
src/configs/kafka.js (7)
12-14: LGTM! Service imports for event processing.The service imports are correctly added to support the event-driven functionality.
25-25: LGTM! Event consumer creation.The event consumer is properly created with a separate group ID, which ensures independent processing of events.
29-29: LGTM! Event consumer connection.The event consumer connection is appropriately established alongside the existing cache consumer.
42-42: LGTM! Topic subscription for events.The event consumer correctly subscribes to the EVENTS_TOPIC from the environment configuration.
44-55: Refactor cache consumer to use concurrent execution.The cache consumer should be refactored to start concurrently with the events consumer rather than sequentially. This change enables both consumers to run in parallel.
56-142: LGTM! Comprehensive event processing implementation.The event consumer implementation effectively addresses the feedback from previous reviews:
✅ Concurrent execution: Both consumers now run concurrently via
Promise.allSettled
✅ Robust error handling: JSON parsing errors are caught and logged appropriately
✅ Payload validation: Required fields are validated before processing
✅ Role diff handling: Role differences are properly computed for any additions or removals
✅ Fallback handling: Delete operations support bothentityIdandidas fallbacks
✅ Non-blocking errors: Errors are logged without crashing the consumerThe implementation correctly handles:
- Create/bulk-create events with proper organization and role mapping
- Delete events with ID validation and fallback logic
- Update/bulk-update events with comprehensive role change detection
- Proper logging with contextual information (topic, partition, offset)
143-143: LGTM! Concurrent consumer execution.The use of
Promise.allSettledensures both consumers run concurrently and continue operating even if one encounters an error, which is the correct approach for long-running Kafka consumers.
|
@coderabbitai full review |
✅ Actions performedFull review triggered. |
There was a problem hiding this comment.
Actionable comments posted: 3
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (6)
src/configs/index.js(1 hunks)src/configs/kafka.js(2 hunks)src/constants/endpoints.js(1 hunks)src/envVariables.js(1 hunks)src/requests/user.js(2 hunks)src/services/users.js(4 hunks)
🧰 Additional context used
🧠 Learnings (2)
📚 Learning: 2025-09-16T10:57:49.662Z
Learnt from: sumanvpacewisdom
PR: ELEVATE-Project/mentoring#1334
File: src/controllers/v1/org-admin.js:117-121
Timestamp: 2025-09-16T10:57:49.662Z
Learning: In the ELEVATE-Project/mentoring repository, internal APIs that are called by other services (like user service) may intentionally accept tenant_code from the request body rather than using req.decodedToken.tenant_code, as this allows the calling service to specify the tenant context for the operation. This is different from external APIs where token-based tenant context is preferred.
Applied to files:
src/requests/user.jssrc/services/users.js
📚 Learning: 2025-09-16T11:06:06.180Z
Learnt from: sumanvpacewisdom
PR: ELEVATE-Project/mentoring#1334
File: src/controllers/v1/entity-type.js:42-44
Timestamp: 2025-09-16T11:06:06.180Z
Learning: The entityTypeService.update method in src/services/entity-type.js has the signature: update(bodyData, id, loggedInUserId, orgCode, tenantCode, roles) - it includes the loggedInUserId parameter between id and orgCode.
Applied to files:
src/requests/user.jssrc/services/users.js
🔇 Additional comments (3)
src/configs/index.js (1)
8-8: Global init order looks good.Executing
globalsbefore the other config loaders keeps shared state ready for Kafka, cache, and bull startup. Looks solid.src/constants/endpoints.js (1)
5-5: Internal profile constant aligns with the new flow.The dedicated constant keeps tenant-aware calls explicit without altering existing paths—nice touch.
src/envVariables.js (1)
571-575: New EVENTS_TOPIC env wiring looks correct.Thanks for registering the topic with a sensible default; this keeps deployments backward-compatible while enabling the new consumer.
| if (message && topic === process.env.CLEAR_INTERNAL_CACHE) { | ||
| if (message.type == 'CLEAR_INTERNAL_CACHE') { | ||
| response = await utils.internalDel(streamingData.value) | ||
| } |
There was a problem hiding this comment.
Fix undefined variable in cache-clear handler
On CLEAR_INTERNAL_CACHE events we call utils.internalDel(streamingData.value), but streamingData is never defined in this scope. Hitting that branch will immediately throw a ReferenceError and keep the cache from clearing. Use the parsed message object instead (and guard for missing value) so the consumer keeps running.
- if (message.type == 'CLEAR_INTERNAL_CACHE') {
- response = await utils.internalDel(streamingData.value)
+ if (message.type === 'CLEAR_INTERNAL_CACHE') {
+ if (!Object.prototype.hasOwnProperty.call(message, 'value')) {
+ logger.warn('CLEAR_INTERNAL_CACHE message missing value; skipping', { topic, partition })
+ return
+ }
+ response = await utils.internalDel(message.value)
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| if (message && topic === process.env.CLEAR_INTERNAL_CACHE) { | |
| if (message.type == 'CLEAR_INTERNAL_CACHE') { | |
| response = await utils.internalDel(streamingData.value) | |
| } | |
| if (message && topic === process.env.CLEAR_INTERNAL_CACHE) { | |
| if (message.type === 'CLEAR_INTERNAL_CACHE') { | |
| if (!Object.prototype.hasOwnProperty.call(message, 'value')) { | |
| logger.warn('CLEAR_INTERNAL_CACHE message missing value; skipping', { topic, partition }) | |
| return | |
| } | |
| response = await utils.internalDel(message.value) | |
| } |
🤖 Prompt for AI Agents
In src/configs/kafka.js around lines 72 to 75, the cache-clear handler
references an undefined variable streamingData, causing a ReferenceError;
replace streamingData.value with the parsed message object (e.g., message.value)
and add a guard to ensure value exists before calling utils.internalDel (return
or log and skip if missing) so the consumer doesn't crash when
CLEAR_INTERNAL_CACHE events lack a value.
| if (err.topics && err.topics[0] === process.env.EVENTS_TOPIC) { | ||
| rolechnageConsumer.errorTriggered(error) | ||
| } | ||
|
|
||
| logger.error(`Error in Kafka message handler for topic ${topic}`, { | ||
| topic, | ||
| partition, | ||
| offset: message.offset, | ||
| err: err?.stack || err?.message || String(err), | ||
| }) |
There was a problem hiding this comment.
Correct typo and variable in error handler
Inside the catch block we call rolechnageConsumer.errorTriggered(error), but rolechnageConsumer is misspelled (should be rolechangeConsumer) and error isn’t defined. That raises another ReferenceError, so even logging fails. Please fix the identifier and forward the original err.
- if (err.topics && err.topics[0] === process.env.EVENTS_TOPIC) {
- rolechnageConsumer.errorTriggered(error)
- }
-
- logger.error(`Error in Kafka message handler for topic ${topic}`, {
- topic,
- partition,
- offset: message.offset,
- err: err?.stack || err?.message || String(err),
- })
+ if (err.topics && err.topics[0] === process.env.EVENTS_TOPIC && rolechangeConsumer?.errorTriggered) {
+ rolechangeConsumer.errorTriggered(err)
+ }Committable suggestion skipped: line range outside the PR's diff.
🤖 Prompt for AI Agents
In src/configs/kafka.js around lines 86 to 95, the catch block mistakenly calls
rolechnageConsumer.errorTriggered(error) which uses a misspelled variable name
and an undefined identifier; change the call to
rolechangeConsumer.errorTriggered(err) so it references the correctly named
consumer and forwards the original caught error (and optionally guard the call
with a truthy check like if (rolechangeConsumer && typeof
rolechangeConsumer.errorTriggered === 'function') before invoking).
|
@coderabbitai full review |
|
@rakeshSgr I'll conduct a full review of the PR changes for event-based user create and role change functionality. ✅ Actions performedFull review triggered. |
✅ Actions performedFull review triggered. |
There was a problem hiding this comment.
Actionable comments posted: 5
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
src/requests/user.js (1)
448-449: ReferenceError:_is not defined._.sortBy is used without importing lodash/underscore; this will crash at runtime.
// Dependencies const userBaseUrl = process.env.USER_SERVICE_HOST + process.env.USER_SERVICE_BASE_URL const requests = require('@generics/requests') const endpoints = require('@constants/endpoints') const request = require('request') const httpStatusCode = require('@generics/http-status') const responses = require('@helpers/responses') const common = require('@constants/common') const { Op } = require('sequelize') +const _ = require('lodash') // needed for _.sortBysrc/configs/kafka.js (1)
1-86: Import missing consumers and avoid message shadowing
- Add imports for
rolechangeConsumer,createuserConsumer,deleteuserConsumer, andupdateuserConsumerat the top ofsrc/configs/kafka.jsto prevent undefined identifier errors.- Do not reassign the Kafka
messageobject; parse into a new variable (e.g.,const parsed = JSON.parse(rawValue)) somessage.offsetremains available in error logging.
🧹 Nitpick comments (9)
src/constants/endpoints.js (1)
5-6: Avoid duplicate constants for the same path.USER_PROFILE_DETAILS_INTERNAL and GET_PROFILE_BY_ID both point to 'v1/user/profileById'. Prefer a single source of truth (reuse GET_PROFILE_BY_ID where possible or add a brief comment clarifying intent) to prevent drift later.
src/envVariables.js (1)
571-575: Name consistency for Kafka topics.Consider aligning with existing naming (e.g., KAFKA_* or *_KAFKA_TOPIC) to keep config uniform across topics.
src/requests/user.js (2)
128-136: Safer URL building for tenantCode and path assembly.Build URLs with URL/URLSearchParams to avoid fragile string concatenation and ensure proper encoding.
-const fetchUserDetails = async ({ token, userId, tenantCode }) => { +const fetchUserDetails = async ({ token, userId, tenantCode }) => { try { - let profileUrl = `${userBaseUrl}${endpoints.USER_PROFILE_DETAILS}` - - if (tenantCode) profileUrl = `${userBaseUrl}${endpoints.USER_PROFILE_DETAILS_INTERNAL}` - - if (userId) profileUrl += `/${userId}` - if (tenantCode) profileUrl += `?tenant_code=${tenantCode}` + const basePath = tenantCode + ? endpoints.USER_PROFILE_DETAILS_INTERNAL + : endpoints.USER_PROFILE_DETAILS + const url = new URL(`${userBaseUrl}${basePath}${userId ? `/${userId}` : ''}`) + if (tenantCode) url.searchParams.set('tenant_code', tenantCode) const isInternalTokenRequired = true - const userDetails = await requests.get(profileUrl, token, isInternalTokenRequired) + const userDetails = await requests.get(url.toString(), token, isInternalTokenRequired)Also update the JSDoc to document tenantCode.
147-157: Defensive role parsing.Guard against non-array/shape mismatches and skip roles without a valid title to prevent runtime errors.
-const rawRoles = userDetails?.data?.result?.user_roles || - userDetails?.data?.result?.organizations?.[0]?.roles || [{ title: common.MENTEE_ROLE }] - -const roles = rawRoles.reduce((acc, role) => { - if (validRoles.has(role.title)) { - if (role.title === common.MENTOR_ROLE) isMentor = true - else if (role.title === common.MENTEE_ROLE) isMenteeRolePresent = true - acc.push(role) - } - return acc -}, []) +const rawRoles = + userDetails?.data?.result?.user_roles ?? + userDetails?.data?.result?.organizations?.[0]?.roles ?? + [{ title: common.MENTEE_ROLE }] +const rolesArray = Array.isArray(rawRoles) ? rawRoles : [{ title: common.MENTEE_ROLE }] +const roles = rolesArray.reduce((acc, role) => { + const title = role?.title + if (typeof title === 'string' && validRoles.has(title)) { + if (title === common.MENTOR_ROLE) isMentor = true + else if (title === common.MENTEE_ROLE) isMenteeRolePresent = true + acc.push(role) + } + return acc +}, [])src/services/users.js (1)
345-345: Docs vs. validation mismatch.validateUserDetails no longer requires email/organization object, but the doc above still lists email and organization as required. Update the JSDoc block for accuracy.
- * - email - * - organizationsrc/configs/kafka.js (4)
74-74: Prefer structured logging and fix typoUse structured fields and correct the label to “Kafka”. Avoid interpolating large objects.
- logger.info(`Kafk event handling response : ${response}`) + logger.info('Kafka event handled', { topic, partition, ok: response !== undefined })
55-67: Unknown event types: log and skip to aid diagnosticsAdd an else branch to warn on unsupported eventType so messages aren’t silently dropped.
if (payload.eventType === 'update' || payload.eventType === 'bulk-update') { response = await updateuserConsumer.messageReceived(payload) } + if (!response) { + logger.warn('Unsupported eventType; skipping', { eventType: payload.eventType }) + }
44-84: Operational hardening: add DLQ/retry and idempotencyConsider routing poison messages to a DLQ topic, adding bounded retries/backoff per handler, and ensuring user/role operations are idempotent (events can be re‑delivered).
1-21: Optional: use KafkaJS instrumentation events instead of string namesKafkaJS exposes producer.events.CONNECT/DISCONNECT; prefer those over string literals if supported in your current version.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (6)
src/configs/index.js(1 hunks)src/configs/kafka.js(1 hunks)src/constants/endpoints.js(1 hunks)src/envVariables.js(1 hunks)src/requests/user.js(2 hunks)src/services/users.js(4 hunks)
🧰 Additional context used
🧠 Learnings (2)
📚 Learning: 2025-09-16T10:57:49.662Z
Learnt from: sumanvpacewisdom
PR: ELEVATE-Project/mentoring#1334
File: src/controllers/v1/org-admin.js:117-121
Timestamp: 2025-09-16T10:57:49.662Z
Learning: In the ELEVATE-Project/mentoring repository, internal APIs that are called by other services (like user service) may intentionally accept tenant_code from the request body rather than using req.decodedToken.tenant_code, as this allows the calling service to specify the tenant context for the operation. This is different from external APIs where token-based tenant context is preferred.
Applied to files:
src/requests/user.js
📚 Learning: 2025-09-16T11:06:06.180Z
Learnt from: sumanvpacewisdom
PR: ELEVATE-Project/mentoring#1334
File: src/controllers/v1/entity-type.js:42-44
Timestamp: 2025-09-16T11:06:06.180Z
Learning: The entityTypeService.update method in src/services/entity-type.js has the signature: update(bodyData, id, loggedInUserId, orgCode, tenantCode, roles) - it includes the loggedInUserId parameter between id and orgCode.
Applied to files:
src/services/users.js
🔇 Additional comments (5)
src/requests/user.js (1)
162-166: Good fix: safe organization_id coercion.Using a null-safe pick plus toString only when present avoids TypeErrors on undefined. Nice.
Based on learnings
src/services/users.js (3)
150-152: Tenant-aware fetch wired correctly.Plumbs tenantCode to fetchUserDetails; matches the internal endpoint change.
194-199: String-coercing IDs is appropriate.toString() on validated ids prevents type drift in persistence and joins.
101-107: Ensure tenantCode is populated in all update() calls
No call sites forupdate()were found with the initial search. Please manually inspect all invocation points (imports, event handlers, controllers) to confirmupdateData.tenantCodeis passed intoupdate().src/configs/kafka.js (1)
69-73: Fix undefined variable in cache-clear path (streamingData) and guardvalueAs previously noted, use the parsed payload and ensure
valueexists before calling internalDel.- if (message && topic === process.env.CLEAR_INTERNAL_CACHE) { - if (message.type === 'CLEAR_INTERNAL_CACHE') { - response = await utils.internalDel(streamingData.value) + if (payload && topic === process.env.CLEAR_INTERNAL_CACHE) { + if (payload.type === 'CLEAR_INTERNAL_CACHE') { + if (!Object.prototype.hasOwnProperty.call(payload, 'value')) { + logger.warn('CLEAR_INTERNAL_CACHE message missing value; skipping', { topic, partition }) + return + } + response = await utils.internalDel(payload.value) } }
| * Description : Contains connections of all configs | ||
| */ | ||
|
|
||
| require('./globals')() |
There was a problem hiding this comment.
🧩 Analysis chain
Confirm globals bootstrap exports a callable and is idempotent.
require('./globals')() assumes module.exports is a function and safe to invoke once at startup. Verify export shape and that repeated requires won’t re-initialize shared state.
🏁 Script executed:
#!/bin/bash
# Check the globals bootstrap shape and idempotency hints
fd -a 'globals.js' src | xargs -I{} sh -c 'echo ">>> {}"; sed -n "1,120p" "{}"'Length of output: 0
🏁 Script executed:
#!/bin/bash
set -e
# Locate globals.js in src/configs and print up to 200 lines
find src/configs -type f -name 'globals.js' -print | while read file; do
echo ">>> $file"
sed -n '1,200p' "$file"
doneLength of output: 0
🏁 Script executed:
#!/bin/bash
set -e
# Locate any file named globals.* in the repo
find . -type f -iname 'globals.*'
echo "-----"
# Show all require('./globals') references with context
rg -n "require\(['\"]\./globals['\"]\)" -g "src/**" -C3Length of output: 272
Verify existence and callable export of globals module
There is no globals.js under src/configs (or anywhere in src); ensure require('./globals') resolves to a module that exports a function and is safe to invoke idempotently.
🤖 Prompt for AI Agents
In src/configs/index.js around line 8, the call require('./globals')() assumes a
./globals module exists and exports a callable function; verify the module path
and either create src/configs/globals.js (or correct relative path) that exports
a function safe to call multiple times, or change the require to the correct
module. Also guard the invocation by requiring the module into a variable and
checking typeof before calling (or wrap in try/catch) so missing or non-callable
exports do not throw at import time.
| await consumer.connect() | ||
| await consumer.subscribe({ topics: [process.env.EVENTS_TOPIC, process.env.CLEAR_INTERNAL_CACHE] }) | ||
|
|
There was a problem hiding this comment.
KafkaJS subscribe API misuse: use topic, not topics
consumer.subscribe expects a single topic per call. Call it twice (or loop), otherwise this will error or be ignored.
- await consumer.subscribe({ topics: [process.env.EVENTS_TOPIC, process.env.CLEAR_INTERNAL_CACHE] })
+ await consumer.subscribe({ topic: process.env.EVENTS_TOPIC })
+ await consumer.subscribe({ topic: process.env.CLEAR_INTERNAL_CACHE })📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| await consumer.connect() | |
| await consumer.subscribe({ topics: [process.env.EVENTS_TOPIC, process.env.CLEAR_INTERNAL_CACHE] }) | |
| await consumer.connect() | |
| await consumer.subscribe({ topic: process.env.EVENTS_TOPIC }) | |
| await consumer.subscribe({ topic: process.env.CLEAR_INTERNAL_CACHE }) |
🤖 Prompt for AI Agents
In src/configs/kafka.js around lines 41 to 43, consumer.subscribe is incorrectly
called with a `topics` array; KafkaJS expects a single `topic` per subscribe
call. Replace the single call with two calls (or loop over the topic list)
invoking await consumer.subscribe({ topic: <topicName> }) for each environment
topic (process.env.EVENTS_TOPIC and process.env.CLEAR_INTERNAL_CACHE) so each
topic is subscribed separately (preserve any existing options like fromBeginning
if needed).
There was a problem hiding this comment.
can be subscribe to multiple topics
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Check KafkaJS version in package.json
cat package.json | jq '.dependencies.kafkajs'
# Search for other KafkaJS subscribe usage in the codebase
rg -n "subscribe.*topics?" --type=js -A2 -B2
# Look for any KafkaJS documentation or examples
fd -e md -e txt | xargs grep -l -i "kafkajs\|kafka" 2>/dev/null | head -5
Length of output: 871
🏁 Script executed:
#!/bin/bash
# Search for package.json files in different locations
find . -name "package.json" -type f | head -3
# Check for yarn.lock or package-lock.json to see KafkaJS version
find . -name "yarn.lock" -o -name "package-lock.json" | head -2
# Look at the KafkaJS import and see if there are any type hints or comments
rg -n "kafkajs" --type=js -A3 -B3
Length of output: 871
🏁 Script executed:
#!/bin/bash
# Check KafkaJS version in src/package.json
cat src/package.json | jq '.dependencies.kafkajs'
# Let's also check if there are any lock files that might show the actual version
find . -name "*lock*" | head -3
# Create a simple test to verify both API approaches work
cat > test_kafka_api.js << 'EOF'
// Simple test to verify KafkaJS subscribe API
const { Kafka } = require('kafkajs');
console.log('Testing KafkaJS subscribe API...');
// Mock consumer to test the API signature
const mockConsumer = {
subscribe: function(options) {
console.log('Subscribe called with:', JSON.stringify(options, null, 2));
if (options.topics && Array.isArray(options.topics)) {
console.log('✓ topics array parameter works');
return Promise.resolve();
}
if (options.topic && typeof options.topic === 'string') {
console.log('✓ topic string parameter works');
return Promise.resolve();
}
console.log('✗ Invalid parameter format');
return Promise.reject(new Error('Invalid subscribe parameters'));
}
};
// Test both approaches
try {
console.log('\n1. Testing topics array:');
mockConsumer.subscribe({ topics: ['topic1', 'topic2'] });
console.log('\n2. Testing single topic:');
mockConsumer.subscribe({ topic: 'topic1' });
} catch (error) {
console.error('Test failed:', error);
}
EOF
node test_kafka_api.js
Length of output: 926
@rakeshSgr You're absolutely correct! KafkaJS does support subscribing to multiple topics using the topics array parameter in the subscribe() method.
Based on the KafkaJS version ^2.2.4 in your package.json and the official documentation, the correct syntax for subscribing to multiple topics is exactly what you have: await consumer.subscribe({ topics: ['topic1', 'topic2'] }).
Your current implementation is correct:
await consumer.subscribe({ topics: [process.env.EVENTS_TOPIC, process.env.CLEAR_INTERNAL_CACHE] })I apologize for the incorrect feedback in the previous comment. The KafkaJS library supports both single topic subscription with { topic: 'topic-name' } and multiple topic subscription with { topics: ['topic1', 'topic2'] }. Your implementation properly subscribes to both topics in a single call, which is the recommended approach.
🌐 Web query:
KafkaJS version 2.2.4 consumer subscribe topics array API documentation
💡 Result:
Short answer — subscribe accepts an array via the topics property.
API (v2.2.4)
- consumer.subscribe({ topic: string }) — subscribe one topic.
- consumer.subscribe({ topics: string[] | RegExp[] , fromBeginning?: boolean, replace?: boolean }) — subscribe multiple topics at once. Example:
await consumer.connect()
await consumer.subscribe({ topics: ['topic-A','topic-B'], fromBeginning: true })
Notes: you may also pass a RegExp in the topics array to match topics. See docs and examples. [1][2][3]
Sources:
[1] KafkaJS docs — Consuming (v2.0.0).
[2] KafkaJS docs — Consuming (v2.1.0).
[3] KafkaJS JSDoc — Consumer.subscribe (describes topic/topics/replace).
@rakeshSgr You're absolutely correct! The web search confirms that KafkaJS v2.2.4 supports subscribing to multiple topics using the topics array parameter.
According to the official KafkaJS documentation for version 2.2.4:
- consumer.subscribe({ topics: string[] | RegExp[] , fromBeginning?: boolean, replace?: boolean }) — subscribe multiple topics at once
Your current implementation is the correct and recommended approach:
await consumer.subscribe({ topics: [process.env.EVENTS_TOPIC, process.env.CLEAR_INTERNAL_CACHE] })I apologize for the incorrect feedback in the previous comment. The KafkaJS library indeed supports both:
- Single topic:
{ topic: 'topic-name' } - Multiple topics:
{ topics: ['topic1', 'topic2'] }
Your code properly subscribes to both topics in a single call, which is more efficient than making separate subscribe calls.
✏️ Learnings added
Learnt from: rakeshSgr
PR: ELEVATE-Project/mentoring#1365
File: src/configs/kafka.js:41-43
Timestamp: 2025-09-29T10:32:14.397Z
Learning: KafkaJS consumer.subscribe() method supports subscribing to multiple topics in a single call using the `topics` array parameter: `await consumer.subscribe({ topics: ['topic1', 'topic2'] })`. This is the correct and recommended approach rather than making separate subscribe calls. KafkaJS version 2.2.4 and later support this API.
Learnt from: rakeshSgr
PR: ELEVATE-Project/mentoring#1365
File: src/configs/kafka.js:41-43
Timestamp: 2025-09-29T10:32:14.397Z
Learning: KafkaJS consumer.subscribe() method supports subscribing to multiple topics in a single call using the `topics` array parameter: `await consumer.subscribe({ topics: ['topic1', 'topic2'] })`. This is the correct and recommended approach rather than making separate subscribe calls. KafkaJS version 2.2.4 and later support this API.
Note: Learnings are effective only in the context of similar code segments. To apply general coding guidelines, please configure review instructions. You can manage existing learnings in the app.
There was a problem hiding this comment.
Actionable comments posted: 3
🧹 Nitpick comments (3)
src/services/users.js (3)
194-199: Centralize ID normalization to reduce type driftCoercing to string is fine; encapsulate it to avoid scattered toString() calls and gracefully handle nulls.
static #getExtensionData(userDetails, orgExtension) { - const data = { - id: userDetails.id.toString(), + const str = (v) => (v == null ? v : String(v)) + const data = { + id: str(userDetails.id), name: userDetails?.name, organization: { - id: orgExtension.organization_id.toString(), + id: str(orgExtension.organization_id), }, }Please confirm downstream query layers accept string IDs (user_id, organization_id). If they expect integers, normalize at boundaries and keep DB layer typed.
326-346: Update JSDoc to match the new required fieldsDocs still list email and organization as required, but validation now checks ['id','user_roles','name','organization_id'].
- * It specifically looks for: - * - id - * - user_roles - * - email - * - name - * - organization - * - organization_id + * It specifically looks for: + * - id + * - user_roles + * - name + * - organization_id
185-189: Return message should reflect create vs updateMinor polish: tailor the success message based on isNewUser for clearer API semantics.
- return responses.successResponse({ - statusCode: httpStatusCode.ok, - message: 'PROFILE_CREATED_SUCCESSFULLY', - result: createOrUpdateResult.result, - }) + const msg = isNewUser + ? 'PROFILE_CREATED_SUCCESSFULLY' + : 'PROFILE_UPDATED_SUCCESSFULLY' + return responses.successResponse({ + statusCode: httpStatusCode.ok, + message: msg, + result: createOrUpdateResult.result, + })
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (6)
src/configs/index.js(1 hunks)src/configs/kafka.js(1 hunks)src/constants/endpoints.js(1 hunks)src/envVariables.js(1 hunks)src/requests/user.js(2 hunks)src/services/users.js(4 hunks)
🧰 Additional context used
🧠 Learnings (2)
📚 Learning: 2025-09-16T10:57:49.662Z
Learnt from: sumanvpacewisdom
PR: ELEVATE-Project/mentoring#1334
File: src/controllers/v1/org-admin.js:117-121
Timestamp: 2025-09-16T10:57:49.662Z
Learning: In the ELEVATE-Project/mentoring repository, internal APIs that are called by other services (like user service) may intentionally accept tenant_code from the request body rather than using req.decodedToken.tenant_code, as this allows the calling service to specify the tenant context for the operation. This is different from external APIs where token-based tenant context is preferred.
Applied to files:
src/requests/user.jssrc/services/users.js
📚 Learning: 2025-09-16T11:06:06.180Z
Learnt from: sumanvpacewisdom
PR: ELEVATE-Project/mentoring#1334
File: src/controllers/v1/entity-type.js:42-44
Timestamp: 2025-09-16T11:06:06.180Z
Learning: The entityTypeService.update method in src/services/entity-type.js has the signature: update(bodyData, id, loggedInUserId, orgCode, tenantCode, roles) - it includes the loggedInUserId parameter between id and orgCode.
Applied to files:
src/requests/user.jssrc/services/users.js
🔇 Additional comments (1)
src/configs/kafka.js (1)
70-72: Fix ReferenceError in CLEAR_INTERNAL_CACHE handler.This branch still calls
utils.internalDel(streamingData.value), butstreamingDatano longer exists—you replaced it withmessage. Every cache-clear event will throw and the cache never gets purged. Switch to the parsed payload (and guard for missingvalue). Same blocker flagged earlier, still unresolved.- if (message.type === 'CLEAR_INTERNAL_CACHE') { - response = await utils.internalDel(streamingData.value) + if (message.type === 'CLEAR_INTERNAL_CACHE') { + if (!Object.prototype.hasOwnProperty.call(message, 'value')) { + logger.warn('CLEAR_INTERNAL_CACHE message missing value; skipping', { topic, partition }) + return + } + response = await utils.internalDel(message.value) }
|
@coderabbitai full review |
✅ Actions performedFull review triggered. |
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
src/services/users.js (2)
114-126: Tenant context is dropped in event “add” pathWhen handling events,
tenantCodeisn’t forwarded to#createOrUpdateUserAndOrg, risking cross‑tenant data access. Pass it through.static async add(bodyData) { bodyData.id = bodyData.id.toString() let result = {} const isNewUser = await this.#checkUserExistence(bodyData.id) if (isNewUser) { - result = await this.#createUserWithBody(bodyData) + result = await this.#createUserWithBody(bodyData) // consider threading tenantCode here too if org ops are tenant-scoped } else { bodyData.new_roles = bodyData.newValues?.organizations?.[0]?.roles ?? [] const targetHasMentorRole = bodyData.new_roles.some((role) => role.title === common.MENTOR_ROLE) - result = await this.#createOrUpdateUserAndOrg(bodyData.id, isNewUser, targetHasMentorRole) + result = await this.#createOrUpdateUserAndOrg( + bodyData.id, + isNewUser, + targetHasMentorRole, + bodyData.tenantCode?.trim?.() || '' + ) } return result }
138-146: Bug: roles overwritten from possibly missinguserBody.roles
#getExtensionDataalready derivesrolesfromuserBody.user_roles; spreading{ roles: userBody.roles }can null/empty it. Keep computed roles or fallback touser_roles.- const createResult = await this.#createUser({ ...userExtensionData, roles: userBody.roles }) + const roles = Array.isArray(userBody.user_roles) ? userBody.user_roles : userExtensionData.roles + const createResult = await this.#createUser({ ...userExtensionData, roles })
🧹 Nitpick comments (8)
src/requests/user.js (1)
147-157: Normalize and de-duplicate roles by title.Current reduce may emit duplicates if both user_roles and organization roles contain the same title. Dedup while preserving mentor/mentee flags.
- const roles = rawRoles.reduce((acc, role) => { - if (validRoles.has(role.title)) { - if (role.title === common.MENTOR_ROLE) isMentor = true - else if (role.title === common.MENTEE_ROLE) isMenteeRolePresent = true - acc.push(role) - } - return acc - }, []) + const roles = [] + const seen = new Set() + for (const role of rawRoles) { + const title = role?.title + if (!title || !validRoles.has(title) || seen.has(title)) continue + if (title === common.MENTOR_ROLE) isMentor = true + else if (title === common.MENTEE_ROLE) isMenteeRolePresent = true + roles.push(role) + seen.add(title) + }src/constants/endpoints.js (1)
5-6: Consolidate duplicate endpoint constants
USER_PROFILE_DETAILS_INTERNAL and GET_PROFILE_BY_ID both map to'v1/user/profileById'. Remove the former and update its usage (src/requests/user.js:132) to use GET_PROFILE_BY_ID, or—if you need a distinct “internal” alias—keep only GET_PROFILE_BY_ID and document its internal use with an inline comment.src/envVariables.js (1)
571-575: EVENTS_TOPIC wiring verified; consider renaming for consistency
- Wiring in
src/configs/kafka.jsconfirmed (subscribe and payload handling useprocess.env.EVENTS_TOPIC).- Optional: rename to
KAFKA_EVENTS_TOPICto align with other{KAFKA_…_TOPIC}env var names.src/configs/kafka.js (3)
36-42: Non‑blocking consumer startup is correctStarting the long‑lived consumer without await prevents init from hanging; error is caught and logged. Consider also wiring a graceful shutdown (SIGTERM) to disconnect producer/consumer.
88-91: Guard missingvaluein CLEAR_INTERNAL_CACHE messagesAvoid calling
internalDelwithundefined. Also accept messages lackingtypewhen already on the cache topic.- if (payload && topic === process.env.CLEAR_INTERNAL_CACHE) { - if (payload.type === 'CLEAR_INTERNAL_CACHE') { - response = await utils.internalDel(payload.value) - } - } + if (payload && topic === process.env.CLEAR_INTERNAL_CACHE) { + if (payload.type === 'CLEAR_INTERNAL_CACHE' || payload.type === undefined) { + if (!Object.prototype.hasOwnProperty.call(payload, 'value')) { + logger.warn('CLEAR_INTERNAL_CACHE message missing value; skipping', { topic, partition, offset }) + } else { + response = await utils.internalDel(payload.value) + } + } + }
93-93: Improve log message (typo + structure)Typo “Kafk” and stringifying objects reduces utility. Prefer structured logging with topic/eventType.
- logger.info(`Kafk event handling response : ${response}`) + logger.info('Kafka event handled', { + topic, + eventType: payload?.eventType, + response, + })src/services/users.js (2)
345-346: Docs mismatch with validationJSDoc above still lists
organizationas required, but code now requires['id','user_roles','name','organization_id']. Update the comment to avoid confusion.
67-70: Use central logger instead ofconsole.login catchesPrefer
logger.error(...)for consistency and observability. Applies to other catches in this file.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (6)
src/configs/index.js(1 hunks)src/configs/kafka.js(2 hunks)src/constants/endpoints.js(1 hunks)src/envVariables.js(1 hunks)src/requests/user.js(2 hunks)src/services/users.js(4 hunks)
🧰 Additional context used
🧠 Learnings (4)
📚 Learning: 2025-09-16T10:57:49.662Z
Learnt from: sumanvpacewisdom
PR: ELEVATE-Project/mentoring#1334
File: src/controllers/v1/org-admin.js:117-121
Timestamp: 2025-09-16T10:57:49.662Z
Learning: In the ELEVATE-Project/mentoring repository, internal APIs that are called by other services (like user service) may intentionally accept tenant_code from the request body rather than using req.decodedToken.tenant_code, as this allows the calling service to specify the tenant context for the operation. This is different from external APIs where token-based tenant context is preferred.
Applied to files:
src/requests/user.jssrc/services/users.js
📚 Learning: 2025-09-16T11:06:06.180Z
Learnt from: sumanvpacewisdom
PR: ELEVATE-Project/mentoring#1334
File: src/controllers/v1/entity-type.js:42-44
Timestamp: 2025-09-16T11:06:06.180Z
Learning: The entityTypeService.update method in src/services/entity-type.js has the signature: update(bodyData, id, loggedInUserId, orgCode, tenantCode, roles) - it includes the loggedInUserId parameter between id and orgCode.
Applied to files:
src/requests/user.jssrc/services/users.js
📚 Learning: 2025-09-29T10:32:14.397Z
Learnt from: rakeshSgr
PR: ELEVATE-Project/mentoring#1365
File: src/configs/kafka.js:41-43
Timestamp: 2025-09-29T10:32:14.397Z
Learning: KafkaJS consumer.subscribe() method supports subscribing to multiple topics in a single call using the `topics` array parameter: `await consumer.subscribe({ topics: ['topic1', 'topic2'] })`. This is the correct and recommended approach rather than making separate subscribe calls. KafkaJS version 2.2.4 and later support this API.
Applied to files:
src/configs/kafka.js
📚 Learning: 2025-08-17T08:50:22.819Z
Learnt from: sumanvpacewisdom
PR: ELEVATE-Project/mentoring#1319
File: src/database/queries/reportTypes.js:13-19
Timestamp: 2025-08-17T08:50:22.819Z
Learning: In multi-tenant database query methods, when accepting an options parameter that gets spread into Sequelize queries, ensure that critical fields like tenant_code are applied after spreading options to prevent bypassing tenant isolation. Use pattern: `const { where: optionsWhere = {}, ...rest } = options || {}; const where = { ...optionsWhere, tenant_code: tenantCode }` to safely merge caller options while enforcing tenant scoping.
Applied to files:
src/services/users.js
🔇 Additional comments (9)
src/requests/user.js (2)
162-166: LGTM: safe string conversion for organization_id.This fixes the earlier .toString() on undefined crash scenario.
137-139: Clarify token forwarding in requests.get
WithisInternalTokenRequired = true,requests.get(url, token, true)attaches both the internal_access_token and the user’s x-auth-token header. Confirm whether this dual-token behavior is intentional for internal endpoints—if not, omit the user token when making internal calls.src/configs/index.js (1)
8-11: Freeze PROJECT_ROOT_DIRECTORY and confirm idempotent bootstraps
- Replace the direct assignment with an immutable definition:
Object.defineProperty(global, 'PROJECT_ROOT_DIRECTORY', { value: path.join(__dirname, '..'), writable: false, configurable: false, });- kafka.js, cache.js, and bull.js already export functions—callable confirmed; ensure repeated
require(...)(…)calls don’t reinitialize connections or recreate resources unexpectedly.src/configs/kafka.js (3)
12-15: Good fix: import previously undefined consumersBrings referenced consumers into scope to avoid runtime ReferenceErrors.
47-49: Correct multi‑topic subscribeUsing
subscribe({ topics: [...] })is the right KafkaJS API for v2.2.x. Based on learnings
60-71: Robust JSON parsing and skip‑on‑errorNice hardening: capture offset, try/catch parse, and continue on invalid JSON with context.
src/services/users.js (3)
150-190: ThreadtenantCodeto downstream operationsYou fetch user details with
tenantCodehere; ensure subsequent role/org updates run in the same tenant (e.g., passtenantCodeintoorgAdminService.roleChange(...), and mentor/mentee create/update paths if they support it).
194-198: Good: normalize IDs to stringsConverting
idandorganization_idto strings prevents type drift across services.
101-107: Validate inputs and normalize types inupdate()Guard against missing
userId, cast to string, and trimtenantCodebefore use. Prevents surprising 404s and undefined lookups.static async update(updateData) { try { - const userId = updateData.userId - - const isNewUser = await this.#checkUserExistence(userId) - const result = await this.#createOrUpdateUserAndOrg(userId, isNewUser, undefined, updateData?.tenantCode) + if (!updateData || !updateData.userId) { + return responses.failureResponse({ + statusCode: httpStatusCode.bad_request, + message: 'INVALID_REQUEST', + responseCode: 'CLIENT_ERROR', + }) + } + const userId = String(updateData.userId) + const tenantCode = + typeof updateData.tenantCode === 'string' ? updateData.tenantCode.trim() : '' + const isNewUser = await this.#checkUserExistence(userId) + const result = await this.#createOrUpdateUserAndOrg(userId, isNewUser, undefined, tenantCode) return result
| const fetchUserDetails = async ({ token, userId, tenantCode }) => { | ||
| try { | ||
| let profileUrl = `${userBaseUrl}${endpoints.USER_PROFILE_DETAILS}` | ||
|
|
||
| if (tenantCode) profileUrl = `${userBaseUrl}${endpoints.USER_PROFILE_DETAILS_INTERNAL}` | ||
|
|
||
| if (userId) profileUrl += `/${userId}` | ||
| if (tenantCode) profileUrl += `?tenant_code=${tenantCode}` | ||
|
|
There was a problem hiding this comment.
Guard against invalid URL when tenantCode is set without userId; also encode tenantCode.
If tenantCode is provided but userId isn’t, the URL becomes /profileById?tenant_code=... which likely 404s. Enforce the input contract and encode the query param.
Apply:
-const fetchUserDetails = async ({ token, userId, tenantCode }) => {
+const fetchUserDetails = async ({ token, userId, tenantCode }) => {
try {
- let profileUrl = `${userBaseUrl}${endpoints.USER_PROFILE_DETAILS}`
-
- if (tenantCode) profileUrl = `${userBaseUrl}${endpoints.USER_PROFILE_DETAILS_INTERNAL}`
-
- if (userId) profileUrl += `/${userId}`
- if (tenantCode) profileUrl += `?tenant_code=${tenantCode}`
+ const isTenantScoped = Boolean(tenantCode)
+ if (isTenantScoped && !userId) {
+ throw new Error('fetchUserDetails: tenantCode requires userId')
+ }
+ let profileUrl = `${userBaseUrl}${
+ isTenantScoped ? endpoints.USER_PROFILE_DETAILS_INTERNAL : endpoints.USER_PROFILE_DETAILS
+ }`
+ if (userId) profileUrl += `/${userId}`
+ if (isTenantScoped) profileUrl += `?tenant_code=${encodeURIComponent(tenantCode)}`Also update the JSDoc to document tenantCode. Based on learnings.
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| const fetchUserDetails = async ({ token, userId, tenantCode }) => { | |
| try { | |
| let profileUrl = `${userBaseUrl}${endpoints.USER_PROFILE_DETAILS}` | |
| if (tenantCode) profileUrl = `${userBaseUrl}${endpoints.USER_PROFILE_DETAILS_INTERNAL}` | |
| if (userId) profileUrl += `/${userId}` | |
| if (tenantCode) profileUrl += `?tenant_code=${tenantCode}` | |
| const fetchUserDetails = async ({ token, userId, tenantCode }) => { | |
| try { | |
| const isTenantScoped = Boolean(tenantCode) | |
| if (isTenantScoped && !userId) { | |
| throw new Error('fetchUserDetails: tenantCode requires userId') | |
| } | |
| let profileUrl = `${userBaseUrl}${ | |
| isTenantScoped | |
| ? endpoints.USER_PROFILE_DETAILS_INTERNAL | |
| : endpoints.USER_PROFILE_DETAILS | |
| }` | |
| if (userId) profileUrl += `/${userId}` | |
| if (isTenantScoped) profileUrl += `?tenant_code=${encodeURIComponent(tenantCode)}` |
🤖 Prompt for AI Agents
In src/requests/user.js around lines 128 to 136, the URL-building logic allows
tenantCode to be appended as a query param even when userId is missing
(producing an invalid /profileById?tenant_code=... URL) and doesn't URL-encode
tenantCode; update the function to validate inputs: if tenantCode is provided
require userId (or throw/return an error), or alternatively call the internal
endpoint only when both tenantCode and userId are present. Ensure tenantCode is
passed through encodeURIComponent when added as a query param, and adjust the
JSDoc to include and document the tenantCode parameter and its
requirement/behavior.
|
|
||
| const isNewUser = await this.#checkUserExistence(userId) | ||
| const result = await this.#createOrUpdateUserAndOrg(userId, isNewUser) | ||
| const result = await this.#createOrUpdateUserAndOrg(userId, isNewUser, undefined, updateData?.tenantCode) |
There was a problem hiding this comment.
@rakeshSgr please add why undefined arguement is passed.
There was a problem hiding this comment.
Its in function declaration it was declared as this
#createOrUpdateUserAndOrg(userId, isNewUser, targetHasMentorRole = undefined) , I have added tenantCode additional parameter
Summary by CodeRabbit
New Features
Bug Fixes
Chores