refactor: completely switch to kafka system

Author: izadoesdev

apps/basket/src/lib/blocked-traffic.ts

@@ -1,9 +1,5 @@
 import { randomUUID } from 'node:crypto';
-import {
-	type AnalyticsEvent,
-	type BlockedTraffic,
-	clickHouse,
-} from '@databuddy/db';
+import { type BlockedTraffic } from '@databuddy/db';
 import { extractIpFromRequest, getGeo } from '../utils/ip-geo';
 import { parseUserAgent } from '../utils/user-agent';
 import { sanitizeString, VALIDATION_LIMITS } from '../utils/validation';
@@ -87,23 +83,11 @@ export async function logBlockedTraffic(
 			created_at: now,
 		};
 
-		clickHouse
-			.insert({
-				table: 'analytics.blocked_traffic',
-				values: [blockedEvent],
-				format: 'JSONEachRow',
-			})
-			.then(() => {
-				if (process.env.ENABLE_KAFKA_EVENTS === 'true') {
-					sendEvent('analytics-blocked-traffic', blockedEvent);
-				}
-			})
-			.catch((err) => {
-				console.error('Failed to log blocked traffic', { error: err as Error });
-				throw err;
-			});
+		sendEvent('analytics-blocked-traffic', blockedEvent);
 	} catch (error) {
-		console.error('Failed to log blocked traffic', { error: error as Error });
+		console.error('Failed to send blocked traffic to Kafka', {
+			error: error as Error,
+		});
 		throw error;
 	}
 }

apps/basket/src/lib/event-service.ts

@@ -5,7 +5,6 @@ import {
 	type CustomOutgoingLink,
 	type ErrorEvent,
 	type WebVitalsEvent,
-	clickHouse,
 } from '@databuddy/db';
 import { checkDuplicate } from './security';
 import { sendEvent } from './producer';
@@ -86,24 +85,9 @@ export async function insertError(
 	};
 
 	try {
-		await clickHouse.insert({
-			table: 'analytics.errors',
-			values: [errorEvent],
-			format: 'JSONEachRow',
-		});
-
-		if (process.env.ENABLE_KAFKA_EVENTS === 'true') {
-			try {
-				sendEvent('analytics-errors', errorEvent);
-			} catch (kafkaErr) {
-				console.error('Failed to send error event to Kafka', {
-					error: kafkaErr as Error,
-					eventId,
-				});
-			}
-		}
+		sendEvent('analytics-errors', errorEvent);
 	} catch (err) {
-		console.error('Failed to insert error event', {
+		console.error('Failed to send error event to Kafka', {
 			error: err as Error,
 			eventId,
 		});
@@ -168,24 +152,9 @@ export async function insertWebVitals(
 	};
 
 	try {
-		await clickHouse.insert({
-			table: 'analytics.web_vitals',
-			values: [webVitalsEvent],
-			format: 'JSONEachRow',
-		});
-
-		if (process.env.ENABLE_KAFKA_EVENTS === 'true') {
-			try {
-				sendEvent('analytics-web-vitals', webVitalsEvent);
-			} catch (kafkaErr) {
-				console.error('Failed to send web vitals event to Kafka', {
-					error: kafkaErr as Error,
-					eventId,
-				});
-			}
-		}
+		sendEvent('analytics-web-vitals', webVitalsEvent);
 	} catch (err) {
-		console.error('Failed to insert web vitals event', {
+		console.error('Failed to send web vitals event to Kafka', {
 			error: err as Error,
 			eventId,
 		});
@@ -237,24 +206,9 @@ export async function insertCustomEvent(
 	};
 
 	try {
-		await clickHouse.insert({
-			table: 'analytics.custom_events',
-			values: [customEvent],
-			format: 'JSONEachRow',
-		});
-
-		if (process.env.ENABLE_KAFKA_EVENTS === 'true') {
-			try {
-				sendEvent('analytics-custom-events', customEvent);
-			} catch (kafkaErr) {
-				console.error('Failed to send custom event to Kafka', {
-					error: kafkaErr as Error,
-					eventId,
-				});
-			}
-		}
+		sendEvent('analytics-custom-events', customEvent);
 	} catch (err) {
-		console.error('Failed to insert custom event', {
+		console.error('Failed to send custom event to Kafka', {
 			error: err as Error,
 			eventId,
 		});
@@ -304,24 +258,9 @@ export async function insertOutgoingLink(
 	};
 
 	try {
-		await clickHouse.insert({
-			table: 'analytics.outgoing_links',
-			values: [outgoingLinkEvent],
-			format: 'JSONEachRow',
-		});
-
-		if (process.env.ENABLE_KAFKA_EVENTS === 'true') {
-			try {
-				sendEvent('analytics-outgoing-links', outgoingLinkEvent);
-			} catch (kafkaErr) {
-				console.error('Failed to send outgoing link event to Kafka', {
-					error: kafkaErr as Error,
-					eventId,
-				});
-			}
-		}
+		sendEvent('analytics-outgoing-links', outgoingLinkEvent);
 	} catch (err) {
-		console.error('Failed to insert outgoing link event', {
+		console.error('Failed to send outgoing link event to Kafka', {
 			error: err as Error,
 			eventId,
 		});
@@ -330,7 +269,7 @@ export async function insertOutgoingLink(
 }
 
 /**
- * Insert a track event (pageview/analytics event) into ClickHouse
+ * Insert a track event (pageview/analytics event) via Kafka
  */
 export async function insertTrackEvent(
 	trackData: any,
@@ -443,24 +382,9 @@ export async function insertTrackEvent(
 	};
 
 	try {
-		await clickHouse.insert({
-			table: 'analytics.events',
-			values: [trackEvent],
-			format: 'JSONEachRow',
-		});
-
-		if (process.env.ENABLE_KAFKA_EVENTS === 'true') {
-			try {
-				sendEvent('analytics-events', trackEvent);
-			} catch (kafkaErr) {
-				console.error('Failed to send track event to Kafka', {
-					error: kafkaErr as Error,
-					eventId,
-				});
-			}
-		}
+		sendEvent('analytics-events', trackEvent);
 	} catch (err) {
-		console.error('Failed to insert track event', {
+		console.error('Failed to send track event to Kafka', {
 			error: err as Error,
 			eventId,
 		});

apps/basket/src/routes/email.ts

@@ -1,5 +1,5 @@
 import { createHash, randomUUID } from 'node:crypto';
-import { clickHouse, type EmailEvent } from '@databuddy/db';
+import { type EmailEvent } from '@databuddy/db';
 import { redis } from '@databuddy/redis';
 import {
 	batchEmailEventSchema,
@@ -44,30 +44,15 @@ async function insertEmailEvent(emailData: EmailEventInput): Promise<void> {
 	};
 
 	try {
-		await clickHouse.insert({
-			table: 'analytics.email_events',
-			values: [emailEvent],
-			format: 'JSONEachRow',
-		});
-
-		if (process.env.ENABLE_KAFKA_EVENTS === 'true') {
-			try {
-				sendEvent('analytics-email-events', emailEvent);
-			} catch (kafkaErr) {
-				logger.error('Failed to send email event to Kafka', {
-					error: kafkaErr as Error,
-					eventId: emailEvent.event_id,
-				});
-			}
-		}
+		sendEvent('analytics-email-events', emailEvent);
 
-		logger.info('Email event inserted successfully', {
+		logger.info('Email event sent to Kafka successfully', {
 			domain: emailEvent.domain,
 			labels: emailEvent.labels,
 			eventId: emailEvent.event_id,
 		});
 	} catch (err) {
-		logger.error('Failed to insert email event', {
+		logger.error('Failed to send email event to Kafka', {
 			error: err as Error,
 			domain: emailEvent.domain,
 			eventId: emailEvent.event_id,