Feat/um 202 event data new (#1841)

* Add event_data base.

* Add url_path.

* Add eventData back.

* Finish event_data relational.

* resolve comments.

lib/clickhouse.ts

@@ -3,6 +3,7 @@ import dateFormat from 'dateformat';
 import debug from 'debug';
 import { FILTER_IGNORED } from 'lib/constants';
 import { CLICKHOUSE } from 'lib/db';
+import { getEventDataType } from './eventData';
 
 export const CLICKHOUSE_DATE_FORMATS = {
   minute: '%Y-%m-%d %H:%M:00',
@@ -64,6 +65,45 @@ function getBetweenDates(field, startAt, endAt) {
   return `${field} between ${getDateFormat(startAt)} and ${getDateFormat(endAt)}`;
 }
 
+function getEventDataFilterQuery(
+  filters: {
+    eventKey?: string;
+    eventValue?: string | number | boolean | Date;
+  }[] = [],
+  params: any,
+) {
+  const query = filters.reduce((ac, cv, i) => {
+    const type = getEventDataType(cv.eventValue);
+
+    let value = cv.eventValue;
+
+    ac.push(`and (event_key = {eventKey${i}:String}`);
+
+    switch (type) {
+      case 'number':
+        ac.push(`and event_numeric_value = {eventValue${i}:UInt64})`);
+        break;
+      case 'string':
+        ac.push(`and event_string_value = {eventValue${i}:String})`);
+        break;
+      case 'boolean':
+        ac.push(`and event_string_value = {eventValue${i}:String})`);
+        value = cv ? 'true' : 'false';
+        break;
+      case 'date':
+        ac.push(`and event_date_value = {eventValue${i}:DateTime('UTC')})`);
+        break;
+    }
+
+    params[`eventKey${i}`] = cv.eventKey;
+    params[`eventValue${i}`] = value;
+
+    return ac;
+  }, []);
+
+  return query.join('\n');
+}
+
 function getFilterQuery(filters = {}, params = {}) {
   const query = Object.keys(filters).reduce((arr, key) => {
     const filter = filters[key];
@@ -189,6 +229,7 @@ export default {
   getDateFormat,
   getBetweenDates,
   getFilterQuery,
+  getEventDataFilterQuery,
   parseFilters,
   findUnique,
   findFirst,

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/eventData.ts

@@ -0,0 +1,74 @@
+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;
+}
+
+export function getEventDataType(value: any): string {
+  let type: string = typeof value;
+
+  if ((type === 'string' && isValid(value)) || isValid(parseISO(value))) {
+    type = 'date';
+  }
+
+  return type;
+}
+
+function createKey(key, value, acc: { keyValues: any[]; parentKey: string }) {
+  const type = getEventDataType(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/prisma.ts

@@ -2,6 +2,7 @@ import prisma from '@umami/prisma-client';
 import moment from 'moment-timezone';
 import { MYSQL, POSTGRESQL, getDatabaseType } from 'lib/db';
 import { FILTER_IGNORED } from 'lib/constants';
+import { getEventDataType } from './eventData';
 
 const MYSQL_DATE_FORMATS = {
   minute: '%Y-%m-%d %H:%i:00',
@@ -64,6 +65,47 @@ function getTimestampInterval(field: string): string {
   }
 }
 
+function getEventDataFilterQuery(
+  filters: {
+    eventKey?: string;
+    eventValue?: string | number | boolean | Date;
+  }[],
+  params: any[],
+) {
+  const query = filters.reduce((ac, cv) => {
+    const type = getEventDataType(cv.eventValue);
+
+    let value = cv.eventValue;
+
+    ac.push(`and (event_key = $${params.length + 1}`);
+    params.push(cv.eventKey);
+
+    switch (type) {
+      case 'number':
+        ac.push(`and event_numeric_value = $${params.length + 1})`);
+        params.push(value);
+        break;
+      case 'string':
+        ac.push(`and event_string_value = $${params.length + 1})`);
+        params.push(decodeURIComponent(cv.eventValue as string));
+        break;
+      case 'boolean':
+        ac.push(`and event_string_value = $${params.length + 1})`);
+        params.push(decodeURIComponent(cv.eventValue as string));
+        value = cv ? 'true' : 'false';
+        break;
+      case 'date':
+        ac.push(`and event_date_value = $${params.length + 1})`);
+        params.push(cv.eventValue);
+        break;
+    }
+
+    return ac;
+  }, []);
+
+  return query.join('\n');
+}
+
 function getFilterQuery(filters = {}, params = []): string {
   const query = Object.keys(filters).reduce((arr, key) => {
     const filter = filters[key];
@@ -173,6 +215,7 @@ export default {
   getDateQuery,
   getTimestampInterval,
   getFilterQuery,
+  getEventDataFilterQuery,
   toUuid,
   parseFilters,
   rawQuery,

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;
@@ -66,6 +75,13 @@ export interface WebsiteEventMetric {
   y: number;
 }
 
+export interface WebsiteEventDataMetric {
+  x: string;
+  t: string;
+  eventName?: string;
+  urlPath?: string;
+}
+
 export interface WebsitePageviews {
   pageviews: {
     t: string;