checkpoint 3

package.json

@@ -58,6 +58,7 @@
     "@commitlint/config-conventional": "^19.2.2",
     "@stryker-mutator/core": "8.2.6",
     "@stryker-mutator/jest-runner": "8.2.6",
+    "@testcontainers/kafka": "^10.9.0",
     "@testcontainers/postgresql": "10.9.0",
     "@testcontainers/redis": "10.9.0",
     "@types/bcrypt": "5.0.2",

src/config/kafka.ts

@@ -1,13 +1,4 @@
-import { Kafka, logLevel, CompressionTypes } from 'kafkajs'
-
-import { getEnv } from './enviroment_config'
-
-const kafka = new Kafka({
-  logLevel: logLevel.ERROR,
-  clientId: getEnv().app.name,
-  connectionTimeout: 5000,
-  brokers: [`${getEnv().broker.url}`],
-})
+import { Kafka, CompressionTypes } from 'kafkajs'
 
 type TOPIC =
   | '2FA_EMAIL_CREATED'
@@ -17,29 +8,12 @@ type TOPIC =
   | 'USER_CREATED'
   | 'ORGANIZATION_CREATED'
   | 'RESET_PASSWORD'
-const TOPIC_LIST: Array<TOPIC> = [
-  '2FA_EMAIL_CREATED',
-  '2FA_PHONE_CREATED',
-  '2FA_EMAIL_SENT',
-  '2FA_PHONE_SENT',
-  'USER_CREATED',
-  'ORGANIZATION_CREATED',
-  'RESET_PASSWORD',
-]
-
-export async function configKafka() {
-  const admin = kafka.admin()
-  await admin.connect()
-  const currentTopic = await admin.listTopics()
-  const toBeCreatedTopic = TOPIC_LIST.filter((t) => !currentTopic.includes(t))
-  await admin.createTopics({
-    waitForLeaders: true,
-    topics: toBeCreatedTopic.map((t) => ({ topic: t })),
-  })
-  await admin.disconnect()
-}
 
-export async function produce<PayloadType>(topic: TOPIC, payload: PayloadType) {
+export async function produce<PayloadType>(
+  topic: TOPIC,
+  payload: PayloadType,
+  kafka: Kafka
+) {
   const producer = kafka.producer()
   await producer.connect()
   await producer.send({

src/config/metric.ts

@@ -6,7 +6,6 @@ import {
 } from 'prom-client'
 
 import { getEnv } from './enviroment_config'
-console.warn(getEnv().app)
 collectDefaultMetrics({ prefix: getEnv().app.name.replaceAll('-', '_') })
 
 export const registry = new Registry()

src/core/index.ts

@@ -1,3 +1,4 @@
+import { Kafka, logLevel } from 'kafkajs'
 import knex from 'knex'
 import { createClient } from 'redis'
 
@@ -71,6 +72,13 @@ export function getCore() {
     })
   })
 
+  const kafka = new Kafka({
+    logLevel: logLevel.ERROR,
+    clientId: getEnv().app.name,
+    connectionTimeout: 5000,
+    brokers: [`${getEnv().broker.url}`],
+  })
+
   // SERVICES
   const passwordService = new PasswordService()
   const uuidService = new UuidService()
@@ -121,13 +129,21 @@ export function getCore() {
   const invalidatingToken: InvalidatingToken = new TokenRepository()
   const creatingToken: CreatingToken = new TokenRepository()
   const decodingToken: DecodingToken = new TokenRepository()
-  const sendingMfaCode: SendingMfaCode = new NotificationProvider(database)
-  const sendingMfaHash: SendingMfaHash = new NotificationProvider(database)
+  const sendingMfaCode: SendingMfaCode = new NotificationProvider(
+    database,
+    kafka
+  )
+  const sendingMfaHash: SendingMfaHash = new NotificationProvider(
+    database,
+    kafka
+  )
   const sendingResetEmail: SendingResetEmail = new NotificationProvider(
-    database
+    database,
+    kafka
   )
   const creatingSystemUser: CreatingSystemUser = new NotificationProvider(
-    database
+    database,
+    kafka
   )
   const creatingResetPassword: CreatingResetPassword =
     new ResetPasswordRepository(uuidService)

src/core/providers/notification.provider.ts

@@ -1,3 +1,4 @@
+import { Kafka } from 'kafkajs'
 import { Knex } from 'knex'
 
 import { produce } from '../../config/kafka'
@@ -22,7 +23,10 @@ export class NotificationProvider
     SendingResetEmail,
     CreatingSystemUser
 {
-  constructor(private database: Knex) {}
+  constructor(
+    private database: Knex,
+    private kafkaClient: Kafka
+  ) {}
 
   async sendCodeByEmail(userId: string, code: string): Promise<void> {
     const tuples: { email: string }[] = await this.database('user')
@@ -34,10 +38,14 @@ export class NotificationProvider
         SendingMfaCodeErrorsTypes.USER_EMAIL_NOT_FOUND
       )
     }
-    await produce('2FA_EMAIL_SENT', {
-      email: tuples[0].email,
-      content: code,
-    })
+    await produce(
+      '2FA_EMAIL_SENT',
+      {
+        email: tuples[0].email,
+        content: code,
+      },
+      this.kafkaClient
+    )
   }
 
   async sendCodeByPhone(userId: string, code: string): Promise<void> {
@@ -52,10 +60,14 @@ export class NotificationProvider
         SendingMfaCodeErrorsTypes.USER_PHONE_NOT_FOUND
       )
     }
-    await produce('2FA_PHONE_SENT', {
-      email: tuples[0].value,
-      content: code,
-    })
+    await produce(
+      '2FA_PHONE_SENT',
+      {
+        email: tuples[0].value,
+        content: code,
+      },
+      this.kafkaClient
+    )
   }
 
   async sendCodeByStrategy(
@@ -85,10 +97,14 @@ export class NotificationProvider
         SendingMfaHashErrorsTypes.USER_EMAIL_HASH_NOT_FOUND
       )
     }
-    await produce('2FA_EMAIL_CREATED', {
-      email: tuples[0].email,
-      content: hash,
-    })
+    await produce(
+      '2FA_EMAIL_CREATED',
+      {
+        email: tuples[0].email,
+        content: hash,
+      },
+      this.kafkaClient
+    )
   }
 
   async sendMfaHashByPhone(userId: string, hash: string): Promise<void> {
@@ -103,10 +119,14 @@ export class NotificationProvider
         SendingMfaHashErrorsTypes.USER_PHONE_HASH_NOT_FOUND
       )
     }
-    await produce('2FA_PHONE_CREATED', {
-      email: tuples[0].value,
-      content: hash,
-    })
+    await produce(
+      '2FA_PHONE_CREATED',
+      {
+        email: tuples[0].value,
+        content: hash,
+      },
+      this.kafkaClient
+    )
   }
 
   async sendMfaHashByStrategy(
@@ -127,15 +147,23 @@ export class NotificationProvider
   }
 
   async sendEmail(email: string, hash: string): Promise<void> {
-    await produce('RESET_PASSWORD', {
-      email: email,
-      hash: hash,
-    })
+    await produce(
+      'RESET_PASSWORD',
+      {
+        email: email,
+        hash: hash,
+      },
+      this.kafkaClient
+    )
   }
 
   async create(userId: string): Promise<void> {
-    await produce('USER_CREATED', {
-      external_id: userId,
-    })
+    await produce(
+      'USER_CREATED',
+      {
+        external_id: userId,
+      },
+      this.kafkaClient
+    )
   }
 }

test/fixtures/multi_factor_authentication.ts

@@ -8,7 +8,7 @@ interface MfaInput {
   isEnable?: boolean
 }
 
-export async function insertMfaIntoDatabase(database: Knex, input?: MfaInput) {
+export async function insertMfaIntoDatabase(database: Knex, input: MfaInput) {
   const row: Array<{ id: string }> = await database(
     'multi_factor_authentication'
   )

test/fixtures/organization.ts

@@ -3,17 +3,15 @@ import { Knex } from 'knex'
 
 interface OrganizationInput {
   name?: string
-  parentOrganizationId?: string
+  parentOrganizationId: string | null
 }
 
 export async function insertOrgIntoDatabase(
   database: Knex,
   input?: OrganizationInput
 ) {
-  let name = ''
-  if (!input?.name) {
-    name = casual.company_name
-  }
+  const name = input?.name ?? casual.company_name
+
   const row: Array<{ id: string }> = await database('organization')
     .insert({
       name,