Adopt old/new event payloads and replace change-sets with extractDelta

src/constants/blacklistConfig.js

@@ -299,6 +299,7 @@ const organization = {
 	update: [
 		'id',
 		'org_admin',
+		'code',
 		'tenant_code',
 		'parent_id',
 		'related_orgs',

src/database/queries/tenants.js

@@ -67,6 +67,7 @@ exports.update = async (filter, updatedata) => {
 		return await Tenant.update(updatedata, {
 			where: filter,
 			returning: true,
+			raw: true,
 		})
 	} catch (error) {
 		throw error

src/distributionColumns.sql

@@ -19,6 +19,7 @@ SELECT create_distributed_table('user_sessions', 'tenant_code');
 -- Organization-related tables
 SELECT create_distributed_table('organization_email_domains', 'tenant_code');
 SELECT create_distributed_table('organization_features', 'tenant_code');
+SELECT create_distributed_table('organization_registration_codes', 'tenant_code');
 
 -- Invitation / user-related
 SELECT create_distributed_table('invitations','tenant_code');

src/dtos/organizationDTO.js

@@ -11,29 +11,20 @@ class organizationDTO {
 		}
 	}
 
-	static eventBodyDTO({ entity, eventType, entityId, changedValues = [], args = {} }) {
+	static eventBodyDTO({ entity, eventType, entityId, oldValues, newValues, args = {} }) {
 		try {
 			if (!entity || !eventType || !entityId)
 				throw new Error('Entity, EventType & EntityId values are mandatory for an Event')
 
 			const disallowedArgs = ['deleted_at']
+			disallowedArgs.forEach((key) => delete args[key])
 
-			disallowedArgs.forEach((key) => {
-				delete args[key]
-			})
-
-			const changes = changedValues.reduce((result, obj) => {
-				const { fieldName, oldValue, newValue } = obj
-				if (!result[fieldName]) result[fieldName] = {}
-				if (oldValue) result[fieldName].oldValue = oldValue
-				if (newValue) result[fieldName].newValue = newValue
-				return result
-			}, {})
 			return {
 				entity,
 				eventType,
 				entityId,
-				changes,
+				...(oldValues ? { oldValues } : {}),
+				...(newValues ? { newValues } : {}),
 				...args,
 			}
 		} catch (error) {

src/dtos/tenantDTO.js

@@ -20,7 +20,7 @@ class TenantResponseDTO {
 		}
 	}
 
-	static eventBodyDTO({ entity, eventType, entityId, changedValues = [], args = {} }) {
+	static eventBodyDTO({ entity, eventType, entityId, oldValues, newValues, args = {} }) {
 		try {
 			if (!entity || !eventType || !entityId)
 				throw new Error('Entity, EventType & EntityId values are mandatory for an Event')
@@ -31,18 +31,12 @@ class TenantResponseDTO {
 				delete args[key]
 			})
 
-			const changes = changedValues.reduce((result, obj) => {
-				const { fieldName, oldValue, newValue } = obj
-				if (!result[fieldName]) result[fieldName] = {}
-				if (oldValue) result[fieldName].oldValue = oldValue
-				if (newValue) result[fieldName].newValue = newValue
-				return result
-			}, {})
 			return {
 				entity,
 				eventType,
 				entityId,
-				changes,
+				...(oldValues ? { oldValues } : {}),
+				...(newValues ? { newValues } : {}),
 				...args,
 			}
 		} catch (error) {

src/dtos/userDTO.js

@@ -77,7 +77,8 @@ class UserDTO {
 		]
 		return input.filter((key) => !restrictedKeys.includes(key))
 	}
-	static eventBodyDTO({ entity, eventType, entityId, changedValues = [], args = {} }) {
+
+	static eventBodyDTO({ entity, eventType, entityId, oldValues = {}, newValues = {}, args = {} }) {
 		try {
 			if (!entity || !eventType || !entityId)
 				throw new Error('Entity, EventType & EntityId values are mandatory for an Event')
@@ -91,23 +92,20 @@ class UserDTO {
 				'custom_entity_text',
 			]
 
-			disallowedArgs.forEach((key) => {
-				delete args[key]
-			})
+			const sanitize = (obj) => {
+				if (obj && typeof obj === 'object') {
+					disallowedArgs.forEach((key) => delete obj[key])
+				}
+				return obj
+			}
 
-			const changes = changedValues.reduce((result, obj) => {
-				const { fieldName, oldValue, newValue } = obj
-				if (!result[fieldName]) result[fieldName] = {}
-				if (oldValue) result[fieldName].oldValue = oldValue
-				if (newValue) result[fieldName].newValue = newValue
-				return result
-			}, {})
 			return {
 				entity,
 				eventType,
 				entityId,
-				changes,
-				...args,
+				...(Object.keys(sanitize(oldValues)).length ? { oldValues } : {}),
+				...(Object.keys(sanitize(newValues)).length ? { newValues } : {}),
+				...sanitize(args),
 			}
 		} catch (error) {
 			console.error(error)

src/generics/utils.js

@@ -1112,6 +1112,29 @@ function extractUpdatedValues(oldData = {}, newData = {}, updateData = {}) {
 	return changes
 }
 
+function flattenLeafPaths(obj, prefix = '') {
+	const out = []
+	for (const [k, v] of Object.entries(obj || {})) {
+		const path = prefix ? `${prefix}.${k}` : k
+		if (v && typeof v === 'object' && !Array.isArray(v)) out.push(...flattenLeafPaths(v, path))
+		else out.push(path)
+	}
+	return out
+}
+
+function extractDelta(oldData = {}, newData = {}, updateData) {
+	const delta = {}
+	const paths =
+		updateData && Object.keys(updateData).length ? flattenLeafPaths(updateData) : flattenLeafPaths(newData) // fallback to full compare
+
+	for (const p of paths) {
+		const oldVal = _.get(oldData, p)
+		const newVal = _.get(newData, p)
+		if (!_.isEqual(oldVal, newVal)) _.set(delta, p, newVal)
+	}
+	return delta
+}
+
 module.exports = {
 	generateToken,
 	hashPassword,
@@ -1159,4 +1182,5 @@ module.exports = {
 	appendParamsToUrl,
 	parseMetaData,
 	extractUpdatedValues,
+	extractDelta,
 }

src/helpers/eventBroadcasterMain.js

@@ -1,6 +1,7 @@
 'use strict'
 const requester = require('@utils/requester')
 const kafkaCommunication = require('@generics/kafka-communication')
+const util = require('util')
 
 const getEndpoints = (eventGroup) => {
 	switch (eventGroup) {
@@ -86,6 +87,9 @@ exports.eventBroadcasterKafka = async (eventGroup, { requestBody }) => {
 }
 exports.broadcastEvent = async (eventGroup, { requestBody, headers = {}, isInternal = true }) => {
 	try {
+		//TODO: Remove this log after testing
+		console.log(util.inspect(requestBody, { depth: null, colors: true, compact: false }))
+
 		// Fire both broadcaster functions concurrently
 		const broadcastPromises = [
 			exports.eventBroadcasterMain(eventGroup, { requestBody, headers, isInternal }),

src/services/admin.js

@@ -722,18 +722,22 @@ module.exports = class AdminHelper {
 
 			//Event body for org update (deactivation)
 			let updatedOrgDetails = orgUpdateResult.updatedRows?.[0]
+
 			const eventBodyData = organizationDTO.eventBodyDTO({
 				entity: 'organization',
-				eventType: 'update',
+				eventType: 'deactivate',
 				entityId: updatedOrgDetails.id,
 				args: {
-					created_by: updatedOrgDetails.created_by,
-					name: updatedOrgDetails.name,
+					id: updatedOrgDetails.id,
 					code: updatedOrgDetails.code,
+					name: updatedOrgDetails.name,
+					created_by: updatedOrgDetails.created_by,
+					updated_by: updatedOrgDetails.updated_by,
 					updated_at: updatedOrgDetails?.updated_at || new Date(),
+					deleted_at: updatedOrgDetails?.deleted_at,
 					status: updatedOrgDetails?.status || common.INACTIVE_STATUS,
-					id: updatedOrgDetails.id,
 					tenant_code: tenantCode,
+					deactivated_users_count: userRowsAffected,
 				},
 			})
 

src/services/org-admin.js

@@ -28,6 +28,8 @@ const emailEncryption = require('@utils/emailEncryption')
 const responses = require('@helpers/responses')
 const notificationUtils = require('@utils/notification')
 const userHelper = require('@helpers/userHelper')
+const { broadcastEvent } = require('@helpers/eventBroadcasterMain')
+const { eventBodyDTO } = require('@dtos/userDTO')
 
 module.exports = class OrgAdminHelper {
 	/**
@@ -628,42 +630,56 @@ module.exports = class OrgAdminHelper {
 	}
 }
 
-function updateRoleForApprovedRequest(requestDetails, user, tenantCode, orgCode) {
-	return new Promise(async (resolve, reject) => {
-		try {
-			const newRole = await roleQueries.findOne(
-				{ id: requestDetails.role, status: common.ACTIVE_STATUS, tenant_code: tenantCode },
-				{ attributes: ['title', 'id', 'user_type', 'status'] }
-			)
+async function updateRoleForApprovedRequest(requestDetails, user, tenantCode, orgCode) {
+	try {
+		const newRole = await roleQueries.findOne(
+			{ id: requestDetails.role, status: common.ACTIVE_STATUS, tenant_code: tenantCode },
+			{ attributes: ['title', 'id', 'user_type', 'status'] }
+		)
 
-			await userOrganizationRoleQueries.create({
-				tenant_code: tenantCode,
-				user_id: user.id,
-				organization_code: orgCode,
-				role_id: newRole.id,
-			})
+		if (!newRole) {
+			throw new Error('ROLE_NOT_FOUND')
+		}
 
-			eventBroadcaster('roleChange', {
-				requestBody: {
-					user_id: requestDetails.requester_id,
-					new_roles: [newRole.title],
-					current_roles: _.map(_.find(user.organizations, { code: orgCode })?.roles || [], 'title'),
-					tenant_code: tenantCode,
-					organization_code: orgCode,
-				},
-			})
-			//delete from cache
-			const redisUserKey = `${common.redisUserPrefix}${tenantCode}_${requestDetails.requester_id.toString()}`
-			await utils.redisDel(redisUserKey)
+		await userOrganizationRoleQueries.create({
+			tenant_code: tenantCode,
+			user_id: user.id,
+			organization_code: orgCode,
+			role_id: newRole.id,
+		})
 
-			return resolve({
-				success: true,
-			})
-		} catch (error) {
-			console.log(error, 'error')
-			return error
-		}
-	})
+		const updatedUser = await userQueries.findUserWithOrganization({
+			id: requestDetails.requester_id,
+			tenant_code: tenantCode,
+		})
+
+		const newValues = utils.extractDelta(user, updatedUser)
+
+		// nothing changed → no event
+		if (Object.keys(newValues).length === 0) return
+
+		const eventBody = eventBodyDTO({
+			entity: 'user',
+			eventType: 'update',
+			entityId: requestDetails.requester_id,
+			oldValues: user,
+			newValues: newValues,
+			args: {
+				created_at: updatedUser.created_at,
+				updated_at: updatedUser.updated_at,
+			},
+		})
+
+		broadcastEvent('userEvents', { requestBody: eventBody, isInternal: true })
+
+		const redisUserKey = `${common.redisUserPrefix}${tenantCode}_${requestDetails.requester_id.toString()}`
+		await utils.redisDel(redisUserKey)
+
+		return { success: true }
+	} catch (error) {
+		console.log(error, 'error')
+		throw error
+	}
 }
 
 async function sendRoleRequestStatusEmail(userDetails, status, organizationCode, tenantCode) {

src/services/organization.js

@@ -830,23 +830,53 @@ async function createRoleRequest(bodyData, tokenInformation) {
 
 async function orgEventEmitter(orgDetailsBeforeUpdate, updatedOrgDetails, bodyData) {
 	// compute changes from provided body keys
-	let changes = await utils.extractUpdatedValues(orgDetailsBeforeUpdate, updatedOrgDetails, bodyData)
-	let related_org_details
-	if (Object.prototype.hasOwnProperty.call(bodyData, 'related_orgs') && updatedOrgDetails?.related_orgs?.length) {
+	const newValues = utils.extractDelta(orgDetailsBeforeUpdate, updatedOrgDetails, bodyData)
+
+	// nothing changed → no event
+	if (Object.keys(newValues).length === 0) return
+
+	// deep-clone old to avoid later mutation bleed
+	const oldValues = _.cloneDeep(orgDetailsBeforeUpdate)
+
+	let related_org_details_new
+	let related_org_details_old
+
+	if (Object.prototype.hasOwnProperty.call(bodyData, 'related_orgs')) {
 		const options = {
 			attributes: ['id', 'code'],
 		}
-		related_org_details = await organizationQueries.findAll(
-			{ id: { [Op.in]: updatedOrgDetails.related_orgs } },
-			options
-		)
+
+		// Handle new values: fetch if non-empty, else set empty array
+		if (updatedOrgDetails?.related_orgs?.length > 0) {
+			related_org_details_new = await organizationQueries.findAll(
+				{ id: { [Op.in]: updatedOrgDetails.related_orgs } },
+				options
+			)
+			newValues.related_org_details = related_org_details_new
+		} else {
+			newValues.related_org_details = []
+		}
+
+		// Handle old values: fetch if non-empty before update, else set empty array
+		// (This covers additions symmetrically, where old was empty)
+		if (orgDetailsBeforeUpdate?.related_orgs?.length > 0) {
+			related_org_details_old = await organizationQueries.findAll(
+				{ id: { [Op.in]: orgDetailsBeforeUpdate.related_orgs } },
+				options
+			)
+			oldValues.related_org_details = related_org_details_old
+		} else {
+			oldValues.related_org_details = []
+		}
 	}
+
 	//event Body for org updates
 	const eventBodyData = organizationDTO.eventBodyDTO({
 		entity: 'organization',
 		eventType: 'update',
 		entityId: orgDetailsBeforeUpdate.id,
-		changedValues: changes,
+		oldValues,
+		newValues,
 		args: {
 			created_by: orgDetailsBeforeUpdate.created_by,
 			name: orgDetailsBeforeUpdate.name,
@@ -860,7 +890,6 @@ async function orgEventEmitter(orgDetailsBeforeUpdate, updatedOrgDetails, bodyDa
 			description: orgDetailsBeforeUpdate.description,
 			related_orgs: orgDetailsBeforeUpdate?.related_orgs || [],
 			tenant_code: orgDetailsBeforeUpdate.tenant_code,
-			...(related_org_details ? { related_org_details } : {}),
 		},
 	})
 	broadcastEvent('organizationEvents', { requestBody: eventBodyData, isInternal: true })