fix: add getReplicatedTableName

packages/db/src/clickhouse/client.ts

@@ -57,6 +57,29 @@ export const TABLE_NAMES = {
   events_imports: 'events_imports',
 };
 
+/**
+ * Check if ClickHouse is running in clustered mode
+ * Clustered mode = production (not self-hosted)
+ * Non-clustered mode = self-hosted environments
+ */
+export function isClickhouseClustered(): boolean {
+  return !(
+    process.env.SELF_HOSTED === 'true' || process.env.SELF_HOSTED === '1'
+  );
+}
+
+/**
+ * Get the replicated table name for mutations
+ * In clustered mode, returns table_name_replicated
+ * In non-clustered mode, returns the original table name
+ */
+export function getReplicatedTableName(tableName: string): string {
+  if (isClickhouseClustered()) {
+    return `${tableName}_replicated ON CLUSTER '{cluster}'`;
+  }
+  return tableName;
+}
+
 export const CLICKHOUSE_OPTIONS: NodeClickHouseClientConfigOptions = {
   max_open_connections: 30,
   request_timeout: 300000,

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

@@ -6,6 +6,7 @@ import {
   chInsertCSV,
   convertClickhouseDateToJs,
   formatClickhouseDate,
+  getReplicatedTableName,
 } from '../clickhouse/client';
 import { csvEscapeField, csvEscapeJson } from '../clickhouse/csv';
 import { type Prisma, db } from '../prisma-client';
@@ -107,8 +108,10 @@ export async function generateSessionIds(
 
   // Use SQL to generate deterministic session IDs based on device_id + 30-min time windows
   // This ensures same events always get same session IDs regardless of import order
+  // In clustered mode, we must use the replicated table for mutations
+  const mutationTableName = getReplicatedTableName(TABLE_NAMES.events_imports);
   const updateQuery = `
-    ALTER TABLE ${TABLE_NAMES.events_imports}
+    ALTER TABLE ${mutationTableName}
     UPDATE session_id = lower(hex(MD5(concat(
       device_id,
       '-',
@@ -572,8 +575,10 @@ export async function backfillSessionsToProduction(
  * Mark import as complete by updating status
  */
 export async function markImportComplete(importId: string): Promise<void> {
+  // In clustered mode, we must use the replicated table for mutations
+  const mutationTableName = getReplicatedTableName(TABLE_NAMES.events_imports);
   const updateQuery = `
-    ALTER TABLE ${TABLE_NAMES.events_imports} 
+    ALTER TABLE ${mutationTableName}
     UPDATE import_status = 'processed'
     WHERE import_id = {importId:String}
   `;