kafka event for tenant

Author: praveenKDass

src/.env.sample

@@ -180,4 +180,10 @@ EVENT_ENABLE_KAFKA_PUSH=true
 #Set Default Phone code 
 DEFAULT_PHONE_CODE="+91"
 # kafka topic for health check
-KAFKA_HEALTH_CHECK_TOPIC='user-health-check-topic-check'   
+KAFKA_HEALTH_CHECK_TOPIC='user-health-check-topic-check'   
+#Enable / disable tenant create event 
+EVENT_ENABLE_TENANT_EVENTS=true 
+#Event Kafka topic for tenant create
+EVENT_TENANT_KAFKA_TOPIC='dev.tenantCreate'
+#Event API for tenant create , can be comma separated
+EVENT_TENANT_LISTENER_API='http://interface:3567/scp/v1/tenant/add'

src/envVariables.js

@@ -415,6 +415,28 @@ let enviromentVariables = {
 		optional: true,
 		default: 'user-health-check-topic-check',
 	},
+	EVENT_ENABLE_TENANT_EVENTS: {
+		message: 'Key to toggle tenant creation kafka event',
+		optional: true,
+	},
+	EVENT_TENANT_KAFKA_TOPIC: {
+		message: 'Kafka topic for Tenant creation Event',
+		optional: true,
+		requiredIf: {
+			key: 'EVENT_ENABLE_TENANT_EVENTS',
+			operator: 'EQUALS',
+			value: 'true',
+		},
+	},
+	EVENT_TENANT_LISTENER_API: {
+		message: 'URL for Tenant creation Event',
+		optional: true,
+		requiredIf: {
+			key: 'EVENT_ENABLE_TENANT_EVENTS',
+			operator: 'EQUALS',
+			value: 'true',
+		},
+	},
 }
 let success = true
 

src/generics/kafka-communication.js

@@ -25,6 +25,16 @@ const pushUserEventsToKafka = async (message) => {
 	}
 }
 
+const pushTenantEventsToKafka = async (message) => {
+	try {
+		const payload = { topic: process.env.EVENT_TENANT_KAFKA_TOPIC, messages: [{ value: JSON.stringify(message) }] }
+		return await pushPayloadToKafka(payload)
+	} catch (error) {
+		console.log(error)
+		return error
+	}
+}
+
 const pushPayloadToKafka = async (payload) => {
 	try {
 		let response = await kafkaProducer.send(payload)
@@ -51,4 +61,5 @@ module.exports = {
 	pushEmailToKafka,
 	clearInternalCache,
 	pushUserEventsToKafka,
+	pushTenantEventsToKafka,
 }

src/helpers/eventBroadcasterMain.js

@@ -11,6 +11,10 @@ const getEndpoints = (eventGroup) => {
 		case 'userEvents':
 			if (process.env.EVENT_USER_LISTENER_API)
 				return process.env.EVENT_USER_LISTENER_API.split(',').filter((url) => url.trim())
+		case 'tenantEvents':
+			if (process.env.EVENT_TENANT_LISTENER_API)
+				return process.env.EVENT_TENANT_LISTENER_API.split(',').filter((url) => url.trim())
+
 		default:
 			return []
 	}
@@ -23,9 +27,10 @@ const isEventEnabled = (eventGroup) => {
 
 		case 'userEvents':
 			return process.env.EVENT_ENABLE_USER_EVENTS !== 'false'
-
 		case 'userEvents-kafka':
 			return process.env.EVENT_ENABLE_KAFKA_PUSH !== 'false'
+		case 'tenantEvents':
+			return process.env.EVENT_ENABLE_TENANT_EVENTS !== 'false'
 		default:
 			return true
 	}
@@ -56,8 +61,21 @@ exports.eventBroadcasterKafka = async (eventGroup, { requestBody }) => {
 		if (!requestBody) throw new Error('Kafka Event Body Generation Failed')
 		if (!isEventEnabled(`${eventGroup}-kafka`))
 			throw new Error(`Kafka Events Not Enabled For The Group "${eventGroup}"`)
-
-		kafkaCommunication.pushUserEventsToKafka(requestBody)
+		//push to kafka based on eventGroup
+		switch (eventGroup) {
+			case 'organizationEvents':
+				kafkaCommunication.pushOrganizationEventsToKafka(requestBody)
+				break
+			case 'userEvents':
+				kafkaCommunication.pushUserEventsToKafka(requestBody)
+				break
+			case 'tenantEvents':
+				kafkaCommunication.pushTenantEventsToKafka(requestBody)
+				break
+			default:
+				console.log('No Kafka Event Group Found')
+				break
+		}
 	} catch (err) {
 		console.log(err)
 	}

src/services/tenant.js

@@ -30,7 +30,8 @@ const utils = require('@generics/utils')
 const _ = require('lodash')
 const responses = require('@helpers/responses')
 const { Op } = require('sequelize')
-
+const { broadcastUserEvent } = require('@helpers/eventBroadcasterMain')
+const UserTransformDTO = require('@dtos/userDTO')
 module.exports = class tenantHelper {
 	/**
 	 * Create Tenant
@@ -371,6 +372,8 @@ module.exports = class tenantHelper {
 						})
 					})
 				}
+				tenantCreateResponse.orgId = defaultOrgCreateResponse?.result?.id
+				tenantCreateResponse.orgCode = defaultOrgCreateResponse?.result?.code
 				// ******* adding default user roles to Default Org CODE ENDS HERE *******
 			} catch (error) {
 				if (rollbackStack.size() > 0) await rollbackStack.execute()
@@ -381,6 +384,29 @@ module.exports = class tenantHelper {
 				})
 			}
 
+			//event Body for tenant create
+			const eventBody = UserTransformDTO.eventBodyDTO({
+				entity: 'tenant',
+				eventType: 'create',
+				entityId: tenantCreateResponse.code,
+				args: {
+					created_by: tenantCreateResponse.created_by,
+					name: tenantCreateResponse.name,
+					code: tenantCreateResponse.code,
+					created_at: tenantCreateResponse?.created_at || new Date(),
+					updated_at: tenantCreateResponse?.updated_at || new Date(),
+					...tenantCreateResponse.meta,
+					status: tenantCreateResponse?.status || common.ACTIVE_STATUS,
+					deleted: false,
+					org_id: tenantCreateResponse.orgId,
+					org_code: tenantCreateResponse.orgCode,
+					description: tenantCreateResponse.description,
+					logo: tenantCreateResponse.logo,
+				},
+			})
+
+			broadcastUserEvent('tenantEvents', { requestBody: eventBody, isInternal: true })
+
 			return responses.successResponse({
 				statusCode: httpStatusCode.accepted,
 				message: 'TENANT_CREATED_SUCCESSFULLY',