custom events table

Author: izadoesdev

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

@@ -2,13 +2,14 @@ import { randomUUID } from "node:crypto";
 import type {
 	AnalyticsEvent,
 	CustomEvent,
+	CustomEventSpan,
 	CustomOutgoingLink,
 	ErrorEvent,
 	ErrorSpanRow,
 	WebVitalsEvent,
 	WebVitalsSpan,
 } from "@databuddy/db";
-import type { ErrorSpan, IndividualVital } from "@databuddy/validation";
+import type { CustomEventSpanInput, ErrorSpan, IndividualVital } from "@databuddy/validation";
 import { getGeo } from "../utils/ip-geo";
 import { parseUserAgent } from "../utils/user-agent";
 import {
@@ -233,6 +234,49 @@ export async function insertCustomEvent(
 	}
 }
 
+/**
+ * Insert lean custom event spans (v2.x format)
+ */
+export async function insertCustomEventSpans(
+	events: CustomEventSpanInput[],
+	clientId: string
+): Promise<void> {
+	if (events.length === 0) {
+		return;
+	}
+
+	const now = Date.now();
+	const spans: CustomEventSpan[] = [];
+
+	for (const event of events) {
+		// Dedup by client+session+path+eventName+timestamp
+		const dedupKey = `custom_${clientId}_${event.sessionId}_${event.path}_${event.eventName}_${event.timestamp}`;
+		const isDuplicate = await checkDuplicate(dedupKey, "custom");
+		if (isDuplicate) {
+			continue;
+		}
+
+		const span: CustomEventSpan = {
+			client_id: clientId,
+			session_id: validateSessionId(event.sessionId),
+			timestamp: typeof event.timestamp === "number" ? event.timestamp : now,
+			path: sanitizeString(event.path, VALIDATION_LIMITS.STRING_MAX_LENGTH),
+			event_name: sanitizeString(event.eventName, VALIDATION_LIMITS.SHORT_STRING_MAX_LENGTH),
+			properties: (event.properties as Record<string, unknown>) ?? {},
+		};
+
+		spans.push(span);
+	}
+
+	if (spans.length > 0) {
+		try {
+			await sendEventBatch("analytics-custom-event-spans", spans);
+		} catch (error) {
+			captureError(error, { count: spans.length });
+		}
+	}
+}
+
 /**
  * Insert an outgoing link click event into the database
  */

apps/basket/src/routes/basket.ts

@@ -7,12 +7,14 @@ import type {
 	WebVitalsEvent,
 } from "@databuddy/db";
 import {
+	batchedCustomEventSpansSchema,
 	batchedErrorsSchema,
 	batchedVitalsSchema,
 } from "@databuddy/validation";
 import { Elysia } from "elysia";
 import {
 	insertCustomEvent,
+	insertCustomEventSpans,
 	insertCustomEventsBatch,
 	insertError,
 	insertErrorSpans,
@@ -414,6 +416,50 @@ const app = new Elysia()
 			return { status: "error", message: "Internal server error" };
 		}
 	})
+	.post("/events", async (context) => {
+		const { body, query, request } = context as {
+			body: unknown;
+			query: Record<string, string>;
+			request: Request;
+		};
+
+		try {
+			const validation = await validateRequest(body, query, request);
+			if ("error" in validation) {
+				return validation.error;
+			}
+
+			const { clientId, userAgent } = validation;
+
+			const parseResult = batchedCustomEventSpansSchema.safeParse(body);
+
+			if (!parseResult.success) {
+				return createSchemaErrorResponse(parseResult.error.issues);
+			}
+
+			const botError = await checkForBot(
+				request,
+				body,
+				query,
+				clientId,
+				userAgent
+			);
+			if (botError) {
+				return botError.error;
+			}
+
+			await insertCustomEventSpans(parseResult.data, clientId);
+
+			return {
+				status: "success",
+				type: "custom_event",
+				count: parseResult.data.length,
+			};
+		} catch (error) {
+			captureError(error, { message: "Error processing custom events" });
+			return { status: "error", message: "Internal server error" };
+		}
+	})
 	.post("/", async (context) => {
 		const { body, query, request } = context as {
 			body: any;

infra/ingest/vector.yaml

@@ -10,6 +10,7 @@ sources:
       - analytics-web-vitals
       - analytics-vitals-spans
       - analytics-custom-events
+      - analytics-custom-event-spans
       - analytics-outgoing-links
       - analytics-email-events
       - analytics-blocked-traffic
@@ -41,6 +42,7 @@ transforms:
       web_vitals: '.topic == "analytics-web-vitals"'
       vitals_spans: '.topic == "analytics-vitals-spans"'
       custom_events: '.topic == "analytics-custom-events"'
+      custom_event_spans: '.topic == "analytics-custom-event-spans"'
       outgoing_links: '.topic == "analytics-outgoing-links"'
       email_events: '.topic == "analytics-email-events"'
       blocked_traffic: '.topic == "analytics-blocked-traffic"'
@@ -174,6 +176,26 @@ sinks:
     encoding:
       timestamp_format: unix_ms
 
+  clickhouse_custom_event_spans:
+    type: clickhouse
+    inputs:
+      - route_analytics.custom_event_spans
+    endpoint: "${CLICKHOUSE_URL}"
+    database: analytics
+    table: custom_event_spans
+    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
+
   clickhouse_outgoing_links:
     type: clickhouse
     inputs:

packages/db/src/clickhouse/schema.ts

@@ -378,6 +378,7 @@ TTL toDateTime(Timestamp) + toIntervalDay(3)
 SETTINGS ttl_only_drop_parts = 1
 `;
 
+// Legacy table - keeping for backwards compatibility
 const CREATE_CUSTOM_EVENTS_TABLE = `
 CREATE TABLE IF NOT EXISTS ${ANALYTICS_DATABASE}.custom_events (
   id UUID,
@@ -394,6 +395,30 @@ ORDER BY (client_id, timestamp, id)
 SETTINGS index_granularity = 8192
 `;
 
+/**
+ * Lean custom event spans table
+ * Uses JSON for flexible metadata
+ */
+const CREATE_CUSTOM_EVENT_SPANS_TABLE = `
+CREATE TABLE IF NOT EXISTS ${ANALYTICS_DATABASE}.custom_event_spans (
+  client_id String CODEC(ZSTD(1)),
+  session_id String CODEC(ZSTD(1)),
+  
+  timestamp DateTime64(3, 'UTC') CODEC(Delta(8), ZSTD(1)),
+  path String CODEC(ZSTD(1)),
+  
+  event_name LowCardinality(String) CODEC(ZSTD(1)),
+  properties JSON CODEC(ZSTD(1)),
+  
+  INDEX idx_session_id session_id TYPE bloom_filter(0.01) GRANULARITY 1,
+  INDEX idx_event_name event_name TYPE bloom_filter(0.01) GRANULARITY 1
+) ENGINE = MergeTree
+PARTITION BY toDate(timestamp)
+ORDER BY (client_id, event_name, path, timestamp)
+TTL toDateTime(timestamp) + INTERVAL 90 DAY
+SETTINGS index_granularity = 8192, ttl_only_drop_parts = 1
+`;
+
 const CREATE_CUSTOM_OUTGOING_LINKS_TABLE = `
 CREATE TABLE IF NOT EXISTS ${ANALYTICS_DATABASE}.outgoing_links (
   id UUID,
@@ -625,6 +650,7 @@ export type OTelLogs = {
 	LogAttributes: Record<string, string>;
 }
 
+// Legacy type - keeping for backwards compatibility
 export type CustomEvent = {
 	id: string;
 	client_id: string;
@@ -635,6 +661,19 @@ export type CustomEvent = {
 	timestamp: number;
 }
 
+/**
+ * Lean custom event span
+ * properties is flexible JSON
+ */
+export type CustomEventSpan = {
+	client_id: string;
+	session_id: string;
+	timestamp: number;
+	path: string;
+	event_name: string;
+	properties: Record<string, unknown>;
+}
+
 export type CustomOutgoingLink = {
 	id: string;
 	client_id: string;
@@ -741,6 +780,7 @@ export async function initClickHouseSchema() {
 			{ name: "blocked_traffic", query: CREATE_BLOCKED_TRAFFIC_TABLE },
 			{ name: "email_events", query: CREATE_EMAIL_EVENTS_TABLE },
 			{ name: "custom_events", query: CREATE_CUSTOM_EVENTS_TABLE },
+			{ name: "custom_event_spans", query: CREATE_CUSTOM_EVENT_SPANS_TABLE },
 			{ name: "outgoing_links", query: CREATE_CUSTOM_OUTGOING_LINKS_TABLE },
 		];
 

packages/validation/src/schemas/custom-events.ts

@@ -1,6 +1,7 @@
 import z from "zod";
 import { VALIDATION_LIMITS } from "../constants";
 
+// Legacy schema
 export const customEventSchema = z.object({
 	eventId: z.string().max(VALIDATION_LIMITS.EVENT_ID_MAX_LENGTH).optional(),
 	name: z.string().min(1).max(VALIDATION_LIMITS.NAME_MAX_LENGTH),
@@ -10,6 +11,19 @@ export const customEventSchema = z.object({
 	properties: z.json().optional().nullable(),
 });
 
+// Lean custom event span schema (v2.x)
+export const customEventSpanSchema = z.object({
+	timestamp: z.number().int(),
+	path: z.string().max(VALIDATION_LIMITS.PATH_MAX_LENGTH),
+	eventName: z.string().min(1).max(VALIDATION_LIMITS.NAME_MAX_LENGTH),
+	sessionId: z.string().max(VALIDATION_LIMITS.SESSION_ID_MAX_LENGTH).nullable().optional(),
+	properties: z.json().optional().nullable(),
+});
+
+export const batchedCustomEventSpansSchema = z.array(customEventSpanSchema).max(VALIDATION_LIMITS.BATCH_MAX_SIZE);
+
+export type CustomEventSpanInput = z.infer<typeof customEventSpanSchema>;
+
 export const outgoingLinkSchema = z.object({
 	eventId: z.string().max(VALIDATION_LIMITS.EVENT_ID_MAX_LENGTH),
 	anonymousId: z.string().nullable().optional(),