add kafka connection

Author: vmaubert

.env.example

@@ -17,6 +17,10 @@ MAILER_USER=
 MAILER_PASSWORD=
 MAILER_PROVIDER=nodemailer
 
+KAFKA_URL=localhost:9092
+KAFKA_TOPIC_DAI=test-topic
+KAFKA_TOPIC_RAI=test-topic
+
 
 INBOX_MAILBOX_NAME=INBOX
 INBOX_SUCCESSBOX_NAME=INBOX.INBOX.Success

package-lock.json

@@ -2,14 +2,16 @@ import { initKnex } from './repositories/db';
 import { initKysely } from './repositories/kysely';
 import { createServer } from './server';
 import { tryToFixResiduesWithUnknownLabel } from './services/imapService/tryToFixUnknownLabels';
+import { createConsumer } from './services/kafkaService';
 import config from './utils/config';
 
 initKnex();
 initKysely(config.databaseUrl);
 createServer().start();
 await tryToFixResiduesWithUnknownLabel();
+const consumer = await createConsumer();
 
 process.on('SIGINT', () => {
-  console.log('\nGracefully shutting down from SIGINT (Ctrl-C)');
-  process.exit();
+  console.log('\nGracefully shutting down from SIGINT...');
+  consumer.disconnect().then(() => process.exit());
 });

server/index.ts

@@ -5,7 +5,7 @@
   "type": "module",
   "private": true,
   "scripts": {
-    "dev": "npm run migrate-latest && node --enable-source-maps --import @swc-node/register/esm-register --watch --inspect ./index.ts",
+    "dev": "npm run migrate-latest && node --enable-source-maps --import @swc-node/register/esm-register --watch --watch-kill-signal SIGINT --inspect ./index.ts",
     "build": "tsc",
     "tsc": "tsc",
     "start": "npm run migrate-latest && node --enable-source-maps --import @swc-node/register/esm-register ./index.ts",
@@ -51,6 +51,7 @@
     "http-proxy-middleware": "^3.0.5",
     "imapflow": "^1.1.1",
     "jsonwebtoken": "^9.0.0",
+    "kafkajs": "^2.2.4",
     "knex": "^2.4.2",
     "knex-stringcase": "^1.5.5",
     "kysely": "^0.27.4",

server/package.json

@@ -14,7 +14,7 @@ import {
 } from './sachaValidator';
 
 const xml = z.string().brand('XML');
-type Xml = z.infer<typeof xml>;
+export type Xml = z.infer<typeof xml>;
 
 export const generateXMLAcquitement = (
   messagesAcquittement: Acquittement['MessageAcquittement'],

server/services/ediSacha/sachaToXML.ts

@@ -0,0 +1,53 @@
+import {
+  Consumer,
+  EachMessagePayload,
+  Kafka,
+  Partitioners,
+  ProducerRecord
+} from 'kafkajs';
+import config from '../utils/config';
+import { Xml } from './ediSacha/sachaToXML';
+
+const kafka = new Kafka({
+  clientId: 'client-id',
+  brokers: [config.kafka.url]
+});
+
+export const createConsumer = async (): Promise<Consumer> => {
+  const consumer = kafka.consumer({
+    groupId: 'consumer-group'
+  });
+
+  try {
+    await consumer.connect();
+    await consumer.subscribe({
+      topic: config.kafka.topicRAI
+    });
+
+    await consumer.run({
+      eachMessage: async (messagePayload: EachMessagePayload) => {
+        const { topic, partition, message } = messagePayload;
+        const prefix = `${topic}[${partition} | ${message.offset}] / ${message.timestamp}`;
+        console.log(`- ${prefix} ${message.key}#${message.value}`);
+      }
+    });
+  } catch (error) {
+    console.log('Error: ', error);
+  }
+
+  return consumer;
+};
+
+export const sendMessage = async (message: Xml): Promise<void> => {
+  const producer = kafka.producer({
+    createPartitioner: Partitioners.DefaultPartitioner
+  });
+
+  const topicMessages: ProducerRecord = {
+    topic: config.kafka.topicDAI,
+    messages: [{ value: message }]
+  };
+  await producer.connect();
+  await producer.send(topicMessages);
+  await producer.disconnect();
+};

server/services/kafkaService.ts

@@ -96,6 +96,11 @@ interface Config {
   m2mBasicToken: string;
   mattermostIncomingWebhook: string | null;
   browserlessUrl: string;
+  kafka: {
+    url: string;
+    topicDAI: string;
+    topicRAI: string;
+  };
 }
 
 const config = convict<Config>({
@@ -357,6 +362,27 @@ const config = convict<Config>({
     sensitive: true,
     nullable: false,
     default: isProduction ? null : 'ws://localhost:3002?token=1234512345'
+  },
+  kafka: {
+    url: {
+      env: 'KAFKA_URL',
+      format: String,
+      sensitive: true,
+      nullable: false,
+      default: isProduction ? null : 'localhost:9092'
+    },
+    topicDAI: {
+      env: 'KAFKA_TOPIC_DAI',
+      format: String,
+      nullable: false,
+      default: isProduction ? null : 'test-topic'
+    },
+    topicRAI: {
+      env: 'KAFKA_TOPIC_RAI',
+      format: String,
+      nullable: false,
+      default: isProduction ? null : 'test-topic'
+    }
   }
 })
   .validate({ allowed: 'strict' })