feat: add producer to all events

izadoesdev · izadoesdev · commit f1e21ae276e6 · 2025-10-22T02:09:28.000+03:00
diff --git a/apps/basket/src/lib/blocked-traffic.ts b/apps/basket/src/lib/blocked-traffic.ts
@@ -7,6 +7,7 @@ import {
 import { extractIpFromRequest, getGeo } from '../utils/ip-geo';
 import { parseUserAgent } from '../utils/user-agent';
 import { sanitizeString, VALIDATION_LIMITS } from '../utils/validation';
+import { sendEvent } from './producer';
 
 /**
  * Log blocked traffic for security and monitoring purposes
@@ -93,7 +94,9 @@ export async function logBlockedTraffic(
 				format: 'JSONEachRow',
 			})
 			.then(() => {
-				// Successfully logged blocked traffic
+				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 });
diff --git a/apps/basket/src/lib/event-service.ts b/apps/basket/src/lib/event-service.ts
@@ -91,6 +91,17 @@ export async function insertError(
 			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,
+				});
+			}
+		}
 	} catch (err) {
 		console.error('Failed to insert error event', {
 			error: err as Error,
@@ -162,6 +173,17 @@ export async function insertWebVitals(
 			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,
+				});
+			}
+		}
 	} catch (err) {
 		console.error('Failed to insert web vitals event', {
 			error: err as Error,
@@ -220,7 +242,22 @@ export async function insertCustomEvent(
 			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,
+				});
+			}
+		}
 	} catch (err) {
+		console.error('Failed to insert custom event', {
+			error: err as Error,
+			eventId,
+		});
 		throw err;
 	}
 }
@@ -272,6 +309,17 @@ export async function insertOutgoingLink(
 			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,
+				});
+			}
+		}
 	} catch (err) {
 		console.error('Failed to insert outgoing link event', {
 			error: err as Error,
@@ -283,7 +331,6 @@ export async function insertOutgoingLink(
 
 /**
  * Insert a track event (pageview/analytics event) into ClickHouse
- * Optionally also sends to Kafka if feature flag is enabled
  */
 export async function insertTrackEvent(
 	trackData: any,
diff --git a/apps/basket/src/lib/producer.ts b/apps/basket/src/lib/producer.ts
@@ -1,65 +1,64 @@
 import { CompressionTypes, Kafka } from 'kafkajs';
 
 const BROKER = process.env.KAFKA_BROKERS as string;
-console.log('BROKER', BROKER);
 
 const kafka = new Kafka({
-  clientId: 'basket',
-  brokers: [BROKER],
+	clientId: 'basket',
+	brokers: [BROKER],
 });
 
 const producer = kafka.producer({
-  allowAutoTopicCreation: true,
+	allowAutoTopicCreation: true,
 });
 
 let connected = false;
 
 const connectProducer = async () => {
-  if (!connected) {
-    await producer.connect();
-    connected = true;
-  }
+	if (!connected) {
+		await producer.connect();
+		connected = true;
+	}
 };
 
 export const sendEventSync = async (topic: string, event: any, key?: string) => {
-  try {
-    await connectProducer();
-    await producer.send({
-      topic,
-      messages: [{ 
-        value: JSON.stringify(event),
-        key: key || event.client_id
-      }],
-      timeout: 10000,
-      compression: CompressionTypes.GZIP,
-    });
-  } catch (err) {
-    console.error('Failed to send event', err);
-    throw err;
-  }
+	try {
+		await connectProducer();
+		await producer.send({
+			topic,
+			messages: [{
+				value: JSON.stringify(event),
+				key: key || event.client_id,
+			}],
+			timeout: 10000,
+			compression: CompressionTypes.GZIP,
+		});
+	} catch (err) {
+		console.error('Failed to send event', err);
+		throw err;
+	}
 };
 
 export const sendEvent = async (topic: string, event: any, key?: string) => {
-  try {
-    await connectProducer();
-    producer.send({
-      topic,
-      messages: [{ 
-        value: JSON.stringify(event),
-        key: key || event.client_id
-      }],
-      timeout: 10000,
-      compression: CompressionTypes.GZIP,
-    });
-  } catch (err) {
-    console.error('Failed to send event', err);
-    throw err;
-  }
+	try {
+		await connectProducer();
+		producer.send({
+			topic,
+			messages: [{
+				value: JSON.stringify(event),
+				key: key || event.client_id,
+			}],
+			timeout: 10000,
+			compression: CompressionTypes.GZIP,
+		});
+	} catch (err) {
+		console.error('Failed to send event', err);
+		throw err;
+	}
 };
 
 export const disconnectProducer = async () => {
-  if (connected) {
-    await producer.disconnect();
-    connected = false;
-  }
+	if (connected) {
+		await producer.disconnect();
+		connected = false;
+	}
 };
diff --git a/apps/basket/src/routes/email.ts b/apps/basket/src/routes/email.ts
@@ -6,6 +6,7 @@ import {
 	type EmailEventInput,
 	emailEventSchema,
 } from '@databuddy/validation';
+import { sendEvent } from '../lib/producer';
 import { Elysia } from 'elysia';
 import { logger } from '../lib/logger';
 
@@ -49,6 +50,17 @@ async function insertEmailEvent(emailData: EmailEventInput): Promise<void> {
 			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,
+				});
+			}
+		}
+
 		logger.info('Email event inserted successfully', {
 			domain: emailEvent.domain,
 			labels: emailEvent.labels,
diff --git a/apps/dashboard/app/layout.tsx b/apps/dashboard/app/layout.tsx
@@ -116,7 +116,7 @@ export default function RootLayout({
 		>
 			<Databuddy
 				apiUrl={
-					isLocalhost ? 'http://localhost:4000' : 'https://basket.databuddy.cc'
+					isLocalhost ? 'https://staging-basket.databuddy.cc' : 'https://basket.databuddy.cc'
 				}
 				clientId={
 					isLocalhost
diff --git a/infra/ingest/vector.yaml b/infra/ingest/vector.yaml
@@ -5,27 +5,97 @@ sources:
     group_id: vector-analytics
     topics:
       - analytics-events
+      - analytics-errors
+      - analytics-web-vitals
+      - analytics-custom-events
+      - analytics-outgoing-links
+      - analytics-email-events
+      - analytics-blocked-traffic
     auto_offset_reset: earliest
     decoding:
       codec: json
+    partition_assignment: cooperative-sticky
 
-sinks:
-  clickhouse:
-    type: clickhouse
+transforms:
+  route_analytics:
+    type: route
     inputs:
       - redpanda
-    endpoint: "${CLICKHOUSE_URL}"
-    database: analytics
+    route:
+      events: '.topic == "analytics-events"'
+      errors: '.topic == "analytics-errors"'
+      web_vitals: '.topic == "analytics-web-vitals"'
+      custom_events: '.topic == "analytics-custom-events"'
+      outgoing_links: '.topic == "analytics-outgoing-links"'
+      email_events: '.topic == "analytics-email-events"'
+      blocked_traffic: '.topic == "analytics-blocked-traffic"'
+
+sink_defaults: &ch_defaults
+  type: clickhouse
+  endpoint: "${CLICKHOUSE_URL}"
+  database: analytics
+  auth:
+    strategy: basic
+    user: "${CLICKHOUSE_USER}"
+    password: "${CLICKHOUSE_PASSWORD}"
+  batch:
+    max_events: 5000
+    max_bytes: 5000000
+    timeout_secs: 5
+  date_time_best_effort: true
+  skip_unknown_fields: true
+  encoding:
+    timestamp_format: unix_ms
+
+sinks:
+  clickhouse_events:
+    <<: *ch_defaults
+    inputs:
+      - route_analytics.events
     table: events
-    auth:
-      strategy: basic
-      user: "${CLICKHOUSE_USER}"
-      password: "${CLICKHOUSE_PASSWORD}"
-    batch:
-      max_events: 1000
-      max_bytes: 5000000
-      timeout_secs: 5
-    date_time_best_effort: true
-    skip_unknown_fields: true
+
+  clickhouse_errors:
+    <<: *ch_defaults
+    inputs:
+      - route_analytics.errors
+    table: errors
+
+  clickhouse_web_vitals:
+    <<: *ch_defaults
+    inputs:
+      - route_analytics.web_vitals
+    table: web_vitals
+
+  clickhouse_custom_events:
+    <<: *ch_defaults
+    inputs:
+      - route_analytics.custom_events
+    table: custom_events
+
+  clickhouse_outgoing_links:
+    <<: *ch_defaults
+    inputs:
+      - route_analytics.outgoing_links
+    table: outgoing_links
+
+  clickhouse_email_events:
+    <<: *ch_defaults
+    inputs:
+      - route_analytics.email_events
+    table: email_events
+
+  clickhouse_blocked_traffic:
+    <<: *ch_defaults
+    inputs:
+      - route_analytics.blocked_traffic
+    table: blocked_traffic
+
+  dead_letters:
+    type: file
+    inputs:
+      - route_analytics._unmatched
+    path: /var/log/vector/dead_letters.log
     encoding:
-      timestamp_format: unix_ms
+      codec: json
+
+

Original file line number	Diff line number	Diff line change
`@@ -116,7 +116,7 @@ export default function RootLayout({`
`116`	`116`	`>`
`117`	`117`	`<Databuddy`
`118`	`118`	`apiUrl={`
`119`		`- isLocalhost ? 'http://localhost:4000' : 'https://basket.databuddy.cc'`
	`119`	`+ isLocalhost ? 'https://staging-basket.databuddy.cc' : 'https://basket.databuddy.cc'`
`120`	`120`	`}`
`121`	`121`	`clientId={`
`122`	`122`	`isLocalhost`