Merge branch 'dev' of https://github.com/umami-software/umami into feat/clickhouse-mv

src/lib/clickhouse.ts

@@ -2,18 +2,19 @@ import { ClickHouseClient, createClient } from '@clickhouse/client';
 import dateFormat from 'dateformat';
 import debug from 'debug';
 import { CLICKHOUSE } from 'lib/db';
-import { QueryFilters, QueryOptions } from './types';
-import { OPERATORS } from './constants';
+import { PageParams, QueryFilters, QueryOptions } from './types';
+import { DEFAULT_PAGE_SIZE, OPERATORS } from './constants';
 import { fetchWebsite } from './load';
 import { maxDate } from './date';
 import { filtersToArray } from './params';
 
 export const CLICKHOUSE_DATE_FORMATS = {
-  minute: '%Y-%m-%d %H:%i:00',
-  hour: '%Y-%m-%d %H:00:00',
-  day: '%Y-%m-%d',
-  month: '%Y-%m-01',
-  year: '%Y-01-01',
+  second: '%Y-%m-%dT%H:%i:%S',
+  minute: '%Y-%m-%dT%H:%i:00',
+  hour: '%Y-%m-%dT%H:00:00',
+  day: '%Y-%m-%dT00:00:00',
+  month: '%Y-%m-01T00:00:00',
+  year: '%Y-01-01T00:00:00',
 };
 
 const log = debug('umami:clickhouse');
@@ -32,7 +33,7 @@ function getClient() {
   } = new URL(process.env.CLICKHOUSE_URL);
 
   const client = createClient({
-    host: `${protocol}//${hostname}:${port}`,
+    url: `${protocol}//${hostname}:${port}`,
     database: pathname.replace('/', ''),
     username: username,
     password,
@@ -47,11 +48,15 @@ function getClient() {
   return client;
 }
 
-function getDateStringQuery(data: any, unit: string | number) {
+function getDateStringSQL(data: any, unit: string | number, timezone?: string) {
+  if (timezone) {
+    return `formatDateTime(${data}, '${CLICKHOUSE_DATE_FORMATS[unit]}', '${timezone}')`;
+  }
+
   return `formatDateTime(${data}, '${CLICKHOUSE_DATE_FORMATS[unit]}')`;
 }
 
-function getDateQuery(field: string, unit: string, timezone?: string) {
+function getDateSQL(field: string, unit: string, timezone?: string) {
   if (timezone) {
     return `date_trunc('${unit}', ${field}, '${timezone}')`;
   }
@@ -95,6 +100,20 @@ function getFilterQuery(filters: QueryFilters = {}, options: QueryOptions = {})
   return query.join('\n');
 }
 
+function getDateQuery(filters: QueryFilters = {}) {
+  const { startDate, endDate } = filters;
+
+  if (startDate) {
+    if (endDate) {
+      return `and created_at between {startDate:DateTime64} and {endDate:DateTime64}`;
+    } else {
+      return `and created_at >= {startDate:DateTime64}`;
+    }
+  }
+
+  return '';
+}
+
 function getFilterParams(filters: QueryFilters = {}) {
   return filtersToArray(filters).reduce((obj, { name, value }) => {
     if (name && value !== undefined) {
@@ -110,6 +129,7 @@ async function parseFilters(websiteId: string, filters: QueryFilters = {}, optio
 
   return {
     filterQuery: getFilterQuery(filters, options),
+    dateQuery: getDateQuery(filters),
     params: {
       ...getFilterParams(filters),
       websiteId,
@@ -119,6 +139,32 @@ async function parseFilters(websiteId: string, filters: QueryFilters = {}, optio
   };
 }
 
+async function pagedQuery(
+  query: string,
+  queryParams: { [key: string]: any },
+  pageParams: PageParams = {},
+) {
+  const { page = 1, pageSize, orderBy, sortDescending = false } = pageParams;
+  const size = +pageSize || DEFAULT_PAGE_SIZE;
+  const offset = +size * (page - 1);
+  const direction = sortDescending ? 'desc' : 'asc';
+
+  const statements = [
+    orderBy && `order by ${orderBy} ${direction}`,
+    +size > 0 && `limit ${+size} offset ${offset}`,
+  ]
+    .filter(n => n)
+    .join('\n');
+
+  const count = await rawQuery(`select count(*) as num from (${query}) t`, queryParams).then(
+    res => res[0].num,
+  );
+
+  const data = await rawQuery(`${query}${statements}`, queryParams);
+
+  return { data, count, page: +page, pageSize: size, orderBy };
+}
+
 async function rawQuery<T = unknown>(
   query: string,
   params: Record<string, unknown> = {},
@@ -136,7 +182,13 @@ async function rawQuery<T = unknown>(
     format: 'JSONEachRow',
   });
 
-  return resultSet.json();
+  return resultSet.json() as T;
+}
+
+async function insert(table: string, values: any[]) {
+  await connect();
+
+  return clickhouse.insert({ table, values, format: 'JSONEachRow' });
 }
 
 async function findUnique(data: any[]) {
@@ -164,12 +216,14 @@ export default {
   client: clickhouse,
   log,
   connect,
-  getDateStringQuery,
-  getDateQuery,
+  getDateStringSQL,
+  getDateSQL,
   getDateFormat,
   getFilterQuery,
   parseFilters,
+  pagedQuery,
   findUnique,
   findFirst,
   rawQuery,
+  insert,
 };

src/lib/date.ts

@@ -292,7 +292,7 @@ export function getDateArray(data: any[], startDate: Date, endDate: Date, unit:
 
   for (let i = 0; i <= n; i++) {
     const t = start(add(startDate, i));
-    const y = data.find(({ x }) => start(getDateFromString(x)).getTime() === t.getTime())?.y || 0;
+    const y = data.find(({ x }) => start(new Date(x)).getTime() === t.getTime())?.y || 0;
 
     arr.push({ x: t, y });
   }

src/lib/detect.ts

@@ -2,9 +2,9 @@ import path from 'path';
 import { getClientIp } from 'request-ip';
 import { browserName, detectOS } from 'detect-browser';
 import isLocalhost from 'is-localhost-ip';
+import ipaddr from 'ipaddr.js';
 import maxmind from 'maxmind';
 import { safeDecodeURIComponent } from 'next-basics';
-
 import {
   DESKTOP_OS,
   MOBILE_OS,
@@ -137,3 +137,31 @@ export async function getClientInfo(req: NextApiRequestCollect) {
 
   return { userAgent, browser, os, ip, country, subdivision1, subdivision2, city, device };
 }
+
+export function hasBlockedIp(req: NextApiRequestCollect) {
+  const ignoreIps = process.env.IGNORE_IP;
+
+  if (ignoreIps) {
+    const ips = [];
+
+    if (ignoreIps) {
+      ips.push(...ignoreIps.split(',').map(n => n.trim()));
+    }
+
+    const clientIp = getIpAddress(req);
+
+    return ips.find(ip => {
+      if (ip === clientIp) return true;
+
+      // CIDR notation
+      if (ip.indexOf('/') > 0) {
+        const addr = ipaddr.parse(clientIp);
+        const range = ipaddr.parseCIDR(ip);
+
+        if (addr.kind() === range[0].kind() && addr.match(range)) return true;
+      }
+    });
+  }
+
+  return false;
+}

src/lib/kafka.ts

@@ -61,8 +61,8 @@ function getDateFormat(date: Date, format?: string): string {
 }
 
 async function sendMessage(
-  message: { [key: string]: string | number },
   topic: string,
+  message: { [key: string]: string | number },
 ): Promise<RecordMetadata[]> {
   await connect();
 
@@ -77,7 +77,7 @@ async function sendMessage(
   });
 }
 
-async function sendMessages(messages: { [key: string]: string | number }[], topic: string) {
+async function sendMessages(topic: string, messages: { [key: string]: string | number }[]) {
   await connect();
 
   await producer.send({

src/lib/prisma.ts

@@ -60,7 +60,7 @@ function getCastColumnQuery(field: string, type: string): string {
   }
 }
 
-function getDateQuery(field: string, unit: string, timezone?: string): string {
+function getDateSQL(field: string, unit: string, timezone?: string): string {
   const db = getDatabaseType();
 
   if (db === POSTGRESQL) {
@@ -81,7 +81,19 @@ function getDateQuery(field: string, unit: string, timezone?: string): string {
   }
 }
 
-function getTimestampDiffQuery(field1: string, field2: string): string {
+export function getTimestampSQL(field: string) {
+  const db = getDatabaseType();
+
+  if (db === POSTGRESQL) {
+    return `floor(extract(epoch from ${field}))`;
+  }
+
+  if (db === MYSQL) {
+    return `UNIX_TIMESTAMP(${field})`;
+  }
+}
+
+function getTimestampDiffSQL(field1: string, field2: string): string {
   const db = getDatabaseType();
 
   if (db === POSTGRESQL) {
@@ -93,7 +105,7 @@ function getTimestampDiffQuery(field1: string, field2: string): string {
   }
 }
 
-function getSearchQuery(column: string): string {
+function getSearchSQL(column: string): string {
   const db = getDatabaseType();
   const like = db === POSTGRESQL ? 'ilike' : 'like';
 
@@ -137,6 +149,20 @@ function getFilterQuery(filters: QueryFilters = {}, options: QueryOptions = {}):
   return query.join('\n');
 }
 
+function getDateQuery(filters: QueryFilters = {}) {
+  const { startDate, endDate } = filters;
+
+  if (startDate) {
+    if (endDate) {
+      return `and website_event.created_at between {{startDate}} and {{endDate}}`;
+    } else {
+      return `and website_event.created_at >= {{startDate}}`;
+    }
+  }
+
+  return '';
+}
+
 function getFilterParams(filters: QueryFilters = {}) {
   return filtersToArray(filters).reduce((obj, { name, operator, value }) => {
     obj[name] = [OPERATORS.contains, OPERATORS.doesNotContain].includes(operator)
@@ -161,6 +187,7 @@ async function parseFilters(
         ? `inner join session on website_event.session_id = session.session_id`
         : '',
     filterQuery: getFilterQuery(filters, options),
+    dateQuery: getDateQuery(filters),
     params: {
       ...getFilterParams(filters),
       websiteId,
@@ -191,8 +218,8 @@ async function rawQuery(sql: string, data: object): Promise<any> {
   return prisma.rawQuery(query, params);
 }
 
-async function pagedQuery<T>(model: string, criteria: T, filters: PageParams) {
-  const { page = 1, pageSize, orderBy, sortDescending = false } = filters || {};
+async function pagedQuery<T>(model: string, criteria: T, pageParams: PageParams) {
+  const { page = 1, pageSize, orderBy, sortDescending = false } = pageParams || {};
   const size = +pageSize || DEFAULT_PAGE_SIZE;
 
   const data = await prisma.client[model].findMany({
@@ -256,11 +283,11 @@ export default {
   getAddIntervalQuery,
   getCastColumnQuery,
   getDayDiffQuery,
-  getDateQuery,
+  getDateSQL,
   getFilterQuery,
   getSearchParameters,
-  getTimestampDiffQuery,
-  getSearchQuery,
+  getTimestampDiffSQL,
+  getSearchSQL,
   getQueryMode,
   pagedQuery,
   parseFilters,