migration commit, revert later

apps/api/Dockerfile

@@ -14,6 +14,9 @@ ENV CLICKHOUSE_PASSWORD=$CLICKHOUSE_PASSWORD
 ARG CLICKHOUSE_URL 
 ENV CLICKHOUSE_URL=$CLICKHOUSE_URL
 
+ARG CLICKHOUSE_URL_NEW 
+ENV CLICKHOUSE_URL_NEW=$CLICKHOUSE_URL_NEW
+
 ARG CLICKHOUSE_USER 
 ENV CLICKHOUSE_USER=$CLICKHOUSE_USER
 

apps/worker/Dockerfile

@@ -14,6 +14,9 @@ ENV CLICKHOUSE_PASSWORD=$CLICKHOUSE_PASSWORD
 ARG CLICKHOUSE_URL 
 ENV CLICKHOUSE_URL=$CLICKHOUSE_URL
 
+ARG CLICKHOUSE_URL_NEW 
+ENV CLICKHOUSE_URL_NEW=$CLICKHOUSE_URL_NEW
+
 ARG CLICKHOUSE_USER 
 ENV CLICKHOUSE_USER=$CLICKHOUSE_USER
 

packages/db/src/clickhouse-client.ts

@@ -12,6 +12,26 @@ export const ch = createClient({
   },
 });
 
+export const chNew = process.env.CLICKHOUSE_URL_NEW
+  ? createClient({
+      url: process.env.CLICKHOUSE_URL_NEW,
+      username: process.env.CLICKHOUSE_USER,
+      password: process.env.CLICKHOUSE_PASSWORD,
+      database: process.env.CLICKHOUSE_DB,
+      max_open_connections: 10,
+      keep_alive: {
+        enabled: true,
+      },
+    })
+  : {
+      query: async () => {
+        return Promise.reject(new Error('Clickhouse URL not configured'));
+      },
+      insert: async () => {
+        return Promise.reject(new Error('Clickhouse URL not configured'));
+      },
+    };
+
 export async function chQueryWithMeta<T extends Record<string, any>>(
   query: string
 ): Promise<ResponseJSON<T>> {

packages/db/src/services/event.service.ts

@@ -9,6 +9,7 @@ import type { IChartEventFilter } from '@openpanel/validation';
 
 import {
   ch,
+  chNew,
   chQuery,
   convertClickhouseDateToJs,
   formatClickhouseDate,
@@ -289,6 +290,20 @@ export async function createEvent(
       date_time_input_format: 'best_effort',
     },
   });
+  try {
+    if (process.env.CLICKHOUSE_URL_NEW) {
+      await chNew.insert({
+        table: 'events',
+        values: [event],
+        format: 'JSONEachRow',
+        clickhouse_settings: {
+          date_time_input_format: 'best_effort',
+        },
+      });
+    }
+  } catch (e) {
+    console.error('Error inserting into new clickhouse', e);
+  }
 
   redisPub.publish('event', superjson.stringify(transformEvent(event)));
   redis.set(

packages/db/src/services/profile.service.ts

@@ -3,7 +3,7 @@ import { escape } from 'sqlstring';
 import { toDots, toObject } from '@openpanel/common';
 import type { IChartEventFilter } from '@openpanel/validation';
 
-import { ch, chQuery } from '../clickhouse-client';
+import { ch, chNew, chQuery } from '../clickhouse-client';
 import { createSqlBuilder } from '../sql-builder';
 import { getEventFiltersWhereClause } from './chart.service';
 
@@ -203,4 +203,34 @@ export async function upsertProfile({
       },
     ],
   });
+
+  try {
+    if (process.env.CLICKHOUSE_URL_NEW) {
+      await chNew.insert({
+        table: 'profiles',
+        format: 'JSONEachRow',
+        clickhouse_settings: {
+          date_time_input_format: 'best_effort',
+        },
+        values: [
+          {
+            id,
+            first_name: firstName ?? profile?.first_name ?? '',
+            last_name: lastName ?? profile?.last_name ?? '',
+            email: email ?? profile?.email ?? '',
+            avatar: avatar ?? profile?.avatar ?? '',
+            properties: toDots({
+              ...(profile?.properties ?? {}),
+              ...(properties ?? {}),
+            }),
+            project_id: projectId ?? profile?.project_id ?? '',
+            created_at: new Date(),
+            is_external: isExternal,
+          },
+        ],
+      });
+    }
+  } catch (e) {
+    console.error('Error inserting into new clickhouse', e);
+  }
 }