umami/lib/db/kafka.js

import { Kafka, logLevel } from 'kafkajs';
import dateFormat from 'dateformat';

export function getKafkaClient() {
  if (!process.env.KAFKA_URL) {
    return null;
  }

  const url = new URL(process.env.KAFKA_URL);
  const brokers = process.env.KAFKA_BROKER.split(',');

  if (url.username.length === 0 && url.password.length === 0) {
    return new Kafka({
      clientId: 'umami',
      brokers: brokers,
      connectionTimeout: 3000,
      logLevel: logLevel.ERROR,
    });
  } else {
    return new Kafka({
      clientId: 'umami',
      brokers: brokers,
      connectionTimeout: 3000,
      ssl: true,
      sasl: {
        mechanism: 'plain',
        username: url.username,
        password: url.password,
      },
    });
  }
}
const kafka = global.kafka || getKafkaClient();
let kafkaProducer = null;

(async () => {
  kafkaProducer = global.kakfaProducer || (await getKafkaProducer());

  if (process.env.NODE_ENV !== 'production') {
    global.kafka = kafka;
    global.kakfaProducer = kafkaProducer;
  }
})();

export { kafka, kafkaProducer };

export async function getKafkaProducer() {
  const producer = kafka.producer();
  await producer.connect();

  return producer;
}

export async function sendKafkaMessage(params, topic) {
  await kafkaProducer.send({
    topic,
    messages: [
      {
        key: 'key',
        value: JSON.stringify(params),
      },
    ],
    acks: 0,
  });
}

export function getDateFormatKafka(date) {
  return dateFormat(date, 'UTC:yyyy-mm-dd HH:MM:ss');
}

export function getKafkaService() {
  const type = process.env.KAFKA_URL && process.env.KAFKA_URL.split(':')[0];

  return type;
}