Add event_data base.

lib/constants.ts

@@ -38,6 +38,19 @@ export const EVENT_TYPE = {
   customEvent: 2,
 } as const;
 
+export const EVENT_DATA_TYPE = {
+  string: 1,
+  number: 2,
+  boolean: 3,
+  date: 4,
+  array: 5,
+} as const;
+
+export const KAFKA_TOPIC = {
+  event: 'event',
+  eventData: 'event_data',
+} as const;
+
 export const ROLES = {
   admin: 'admin',
   user: 'user',

lib/detect.js

@@ -79,10 +79,10 @@ export async function getClientInfo(req, { screen }) {
   const userAgent = req.headers['user-agent'];
   const ip = getIpAddress(req);
   const location = await getLocation(ip);
-  const country = location.country;
-  const subdivision1 = location.subdivision1;
-  const subdivision2 = location.subdivision2;
-  const city = location.city;
+  const country = location?.country;
+  const subdivision1 = location?.subdivision1;
+  const subdivision2 = location?.subdivision2;
+  const city = location?.city;
   const browser = browserName(userAgent);
   const os = detectOS(userAgent);
   const device = getDevice(screen, browser, os);

lib/eventData.ts

@@ -0,0 +1,64 @@
+import { isValid, parseISO } from 'date-fns';
+import { EVENT_DATA_TYPE } from './constants';
+import { EventDataTypes } from './types';
+
+export function flattenJSON(
+  eventData: { [key: string]: any },
+  keyValues: { key: string; value: any; eventDataType: EventDataTypes }[] = [],
+  parentKey = '',
+): { key: string; value: any; eventDataType: EventDataTypes }[] {
+  return Object.keys(eventData).reduce(
+    (acc, key) => {
+      const value = eventData[key];
+      const type = typeof eventData[key];
+
+      // nested object
+      if (value && type === 'object' && !Array.isArray(value) && !isValid(value)) {
+        flattenJSON(value, acc.keyValues, getKeyName(key, parentKey));
+      } else {
+        createKey(getKeyName(key, parentKey), value, acc);
+      }
+
+      return acc;
+    },
+    { keyValues, parentKey },
+  ).keyValues;
+}
+
+function createKey(key, value, acc: { keyValues: any[]; parentKey: string }) {
+  const type = isValid(value) || isValid(parseISO(value)) ? 'date' : typeof value;
+
+  let eventDataType = null;
+
+  switch (type) {
+    case 'number':
+      eventDataType = EVENT_DATA_TYPE.number;
+      break;
+    case 'string':
+      eventDataType = EVENT_DATA_TYPE.string;
+      break;
+    case 'boolean':
+      eventDataType = EVENT_DATA_TYPE.boolean;
+      break;
+    case 'date':
+      eventDataType = EVENT_DATA_TYPE.date;
+      break;
+    case 'object':
+      eventDataType = EVENT_DATA_TYPE.array;
+      value = JSON.stringify(value);
+      break;
+    default:
+      eventDataType = EVENT_DATA_TYPE.string;
+      break;
+  }
+
+  acc.keyValues.push({ key, value, eventDataType });
+}
+
+function getKeyName(key, parentKey) {
+  if (!parentKey) {
+    return key;
+  }
+
+  return `${parentKey}.${key}`;
+}

lib/kafka.ts

@@ -1,19 +1,20 @@
-import { Kafka, logLevel } from 'kafkajs';
 import dateFormat from 'dateformat';
 import debug from 'debug';
+import { Kafka, Mechanism, Producer, RecordMetadata, SASLOptions, logLevel } from 'kafkajs';
 import { KAFKA, KAFKA_PRODUCER } from 'lib/db';
+import * as tls from 'tls';
 
 const log = debug('umami:kafka');
 
-let kafka;
-let producer;
+let kafka: Kafka;
+let producer: Producer;
 const enabled = Boolean(process.env.KAFKA_URL && process.env.KAFKA_BROKER);
 
 function getClient() {
   const { username, password } = new URL(process.env.KAFKA_URL);
   const brokers = process.env.KAFKA_BROKER.split(',');
 
-  const ssl =
+  const ssl: { ssl?: tls.ConnectionOptions | boolean; sasl?: SASLOptions | Mechanism } =
     username && password
       ? {
           ssl: {
@@ -30,7 +31,7 @@ function getClient() {
         }
       : {};
 
-  const client = new Kafka({
+  const client: Kafka = new Kafka({
     clientId: 'umami',
     brokers: brokers,
     connectionTimeout: 3000,
@@ -47,7 +48,7 @@ function getClient() {
   return client;
 }
 
-async function getProducer() {
+async function getProducer(): Promise<Producer> {
   const producer = kafka.producer();
   await producer.connect();
 
@@ -60,25 +61,40 @@ async function getProducer() {
   return producer;
 }
 
-function getDateFormat(date) {
+function getDateFormat(date): string {
   return dateFormat(date, 'UTC:yyyy-mm-dd HH:MM:ss');
 }
 
-async function sendMessage(params, topic) {
+async function sendMessage(
+  message: { [key: string]: string | number },
+  topic: string,
+): Promise<RecordMetadata[]> {
+  await connect();
+
+  return producer.send({
+    topic,
+    messages: [
+      {
+        value: JSON.stringify(message),
+      },
+    ],
+    acks: -1,
+  });
+}
+
+async function sendMessages(messages: { [key: string]: string | number }[], topic: string) {
   await connect();
 
   await producer.send({
     topic,
-    messages: [
-      {
-        value: JSON.stringify(params),
-      },
-    ],
+    messages: messages.map(a => {
+      return { value: JSON.stringify(a) };
+    }),
     acks: 1,
   });
 }
 
-async function connect() {
+async function connect(): Promise<Kafka> {
   if (!kafka) {
     kafka = process.env.KAFKA_URL && process.env.KAFKA_BROKER && (global[KAFKA] || getClient());
 
@@ -98,4 +114,5 @@ export default {
   connect,
   getDateFormat,
   sendMessage,
+  sendMessages,
 };

lib/types.ts

@@ -1,10 +1,19 @@
 import { NextApiRequest } from 'next';
-import { ROLES } from './constants';
+import { EVENT_DATA_TYPE, EVENT_TYPE, KAFKA_TOPIC, ROLES } from './constants';
 
 type ObjectValues<T> = T[keyof T];
 
 export type Roles = ObjectValues<typeof ROLES>;
 
+export type EventTypes = ObjectValues<typeof EVENT_TYPE>;
+
+export type EventDataTypes = ObjectValues<typeof EVENT_DATA_TYPE>;
+
+export type KafkaTopics = ObjectValues<typeof KAFKA_TOPIC>;
+
+export interface EventData {
+  [key: string]: number | string | EventData | number[] | string[] | EventData[];
+}
 export interface Auth {
   user?: {
     id: string;