Merge pull request #538 from databuddy-analytics/staging

Trait history, billing lifecycle events, and ingestion credential scrubbing

Author: izadoesdev

.env.example

@@ -87,3 +87,7 @@ NOTRA_API_KEY=
 # Superlog OTLP logs export. When set, evlog drains in api/basket/insights/slack/uptime
 # also ship wide events to https://intake.superlog.sh/v1/logs.
 SUPERLOG_API_KEY=""
+
+# Website id used for dogfooding our own product analytics (billing lifecycle
+# events and plan traits written server-side). Leave empty to disable.
+SELF_ANALYTICS_WEBSITE_ID=""

.github/workflows/ci.yml

@@ -111,6 +111,19 @@ jobs:
           --health-interval 10s
           --health-timeout 5s
           --health-retries 5
+      clickhouse:
+        image: clickhouse/clickhouse-server:25.5.1-alpine
+        env:
+          CLICKHOUSE_DB: databuddy_analytics
+          CLICKHOUSE_USER: default
+          CLICKHOUSE_DEFAULT_ACCESS_MANAGEMENT: 1
+        ports:
+          - 8123:8123
+        options: >-
+          --health-cmd "clickhouse-client --query 'SELECT 1'"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
 
     steps:
       - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
@@ -132,6 +145,10 @@ jobs:
       - run: bun install --frozen-lockfile --ignore-scripts
       - name: Push test schema
         run: cd packages/db && bunx drizzle-kit push
+      - name: Init ClickHouse schema
+        env:
+          CLICKHOUSE_URL: http://default:@localhost:8123
+        run: bun run --cwd packages/db clickhouse:init
       - name: Test
         env:
           NODE_ENV: test

apps/api/package.json

@@ -23,6 +23,7 @@
     "@databuddy/redis": "workspace:*",
     "@databuddy/rpc": "workspace:*",
     "@databuddy/sdk": "workspace:*",
+    "@databuddy/services": "workspace:*",
     "@databuddy/shared": "workspace:*",
     "@databuddy/validation": "workspace:*",
     "@elysiajs/cors": "^1.4.1",

apps/api/src/routes/query.ts

@@ -7,8 +7,10 @@ import {
 	hasKeyScope,
 	isApiKeyPresent,
 } from "@databuddy/api-keys/resolve";
-import { db } from "@databuddy/db";
+import { and, db, eq, inArray } from "@databuddy/db";
+import { profiles } from "@databuddy/db/schema";
 import { config } from "@databuddy/env/app";
+import { revealPii } from "@databuddy/services/identity";
 import { validateTimezone } from "@databuddy/validation";
 import { readBooleanEnv } from "@databuddy/env/boolean";
 import { ratelimit } from "@databuddy/redis/rate-limit";
@@ -880,6 +882,53 @@ interface QueryResult {
 	success: boolean;
 }
 
+const PROFILE_ENRICHED_TYPES = new Set(["profile_list"]);
+
+async function attachProfileIdentities(
+	websiteId: string,
+	results: QueryResult[]
+) {
+	const profileIds = new Set<string>();
+	for (const result of results) {
+		for (const row of result.data) {
+			if (typeof row.profile_id === "string" && row.profile_id) {
+				profileIds.add(row.profile_id);
+			}
+		}
+	}
+	if (profileIds.size === 0) {
+		return;
+	}
+
+	const identities = await db
+		.select({
+			profileId: profiles.profileId,
+			displayName: profiles.displayName,
+			email: profiles.email,
+		})
+		.from(profiles)
+		.where(
+			and(
+				eq(profiles.websiteId, websiteId),
+				inArray(profiles.profileId, [...profileIds])
+			)
+		);
+	const identityByProfileId = new Map(
+		identities.map((identity) => [identity.profileId, identity])
+	);
+
+	for (const result of results) {
+		for (const row of result.data) {
+			const identity =
+				typeof row.profile_id === "string"
+					? identityByProfileId.get(row.profile_id)
+					: undefined;
+			row.display_name = identity ? revealPii(identity.displayName) : null;
+			row.email = identity ? revealPii(identity.email) : null;
+		}
+	}
+}
+
 async function executeDynamicQuery(
 	request: DynamicQueryRequestType,
 	projectId: string,
@@ -1026,6 +1075,22 @@ async function executeDynamicQuery(
 				});
 			}
 		}
+
+		if (projectType === "website") {
+			const profileResults = validParameters
+				.filter((p) => PROFILE_ENRICHED_TYPES.has(p.request.type))
+				.flatMap((p) => resultMap.get(p.id) ?? []);
+			if (profileResults.length > 0) {
+				await attachProfileIdentities(projectId, profileResults).catch(
+					(error) => {
+						captureError(error, {
+							route: "v1/query",
+							step: "attach_profile_identities",
+						});
+					}
+				);
+			}
+		}
 	}
 
 	const allResults = prepared.map(

apps/api/src/routes/webhooks/autumn.test.ts

@@ -92,6 +92,10 @@ vi.mock("@databuddy/redis", () => ({
 	invalidateBillingOwnerCaches: vi.fn(async () => ({ attempted: 0, failed: 0 })),
 }));
 
+vi.mock("@databuddy/services/billing-lifecycle", () => ({
+	recordPlanChange: vi.fn(async () => undefined),
+}));
+
 vi.mock("elysia", () => ({
 	Elysia: class {
 		post() {

apps/api/src/routes/webhooks/autumn.ts

@@ -17,6 +17,7 @@ import {
 	invalidateAgentContextSnapshotsForOwner,
 	invalidateBillingOwnerCaches,
 } from "@databuddy/redis";
+import { recordPlanChange } from "@databuddy/services/billing-lifecycle";
 import { Elysia } from "elysia";
 import { useLogger } from "evlog/elysia";
 import { Resend } from "resend";
@@ -324,9 +325,24 @@ async function handleProductsUpdated(
 	});
 	await invalidatePlanCaches(customer.id);
 
-	const shouldSkipSlack =
-		!slack ||
-		(process.env.NODE_ENV === "production" && customer.env === "sandbox");
+	const isSandboxInProduction =
+		process.env.NODE_ENV === "production" && customer.env === "sandbox";
+
+	if (customer.id && !isSandboxInProduction) {
+		try {
+			await recordPlanChange({
+				customerId: customer.id,
+				planId: updated_product.id,
+				scenario,
+			});
+		} catch (error) {
+			log.error(error instanceof Error ? error : new Error(String(error)), {
+				autumn: { step: "record_plan_change", customerId: customer.id },
+			});
+		}
+	}
+
+	const shouldSkipSlack = !slack || isSandboxInProduction;
 
 	if (shouldSkipSlack) {
 		return { success: true, message: `Processed ${scenario}` };

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

@@ -7,7 +7,11 @@ import { runFork, send } from "@lib/producer";
 import { captureError } from "@lib/tracing";
 import { extractIpFromRequest, getGeo } from "@utils/ip-geo";
 import { parseUserAgent } from "@utils/user-agent";
-import { sanitizeString, VALIDATION_LIMITS } from "@utils/validation";
+import {
+	sanitizeString,
+	sanitizeUrl,
+	VALIDATION_LIMITS,
+} from "@utils/validation";
 import { randomUUIDv7 } from "bun";
 
 async function _logBlockedTrafficAsync(
@@ -41,12 +45,12 @@ async function _logBlockedTrafficAsync(
 			client_id: clientId || "",
 			timestamp: now,
 
-			path: sanitizeString(body?.path, VALIDATION_LIMITS.STRING_MAX_LENGTH),
-			url: sanitizeString(
+			path: sanitizeUrl(body?.path, VALIDATION_LIMITS.STRING_MAX_LENGTH),
+			url: sanitizeUrl(
 				body?.url || body?.href,
 				VALIDATION_LIMITS.STRING_MAX_LENGTH
 			),
-			referrer: sanitizeString(
+			referrer: sanitizeUrl(
 				body?.referrer || request.headers.get("referer"),
 				VALIDATION_LIMITS.STRING_MAX_LENGTH
 			),

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

@@ -17,6 +17,7 @@ import { extractTrustedClientIp, getGeo } from "@utils/ip-geo";
 import { parseUserAgent } from "@utils/user-agent";
 import {
 	sanitizeString,
+	sanitizeUrl,
 	VALIDATION_LIMITS,
 	validatePerformanceMetric,
 	validateSessionId,
@@ -77,12 +78,12 @@ export function buildTrackEvent(
 		event_id: ctx.eventId,
 		session_start_time: sessionStartTime,
 		timestamp,
-		referrer: sanitizeString(
+		referrer: sanitizeUrl(
 			trackData.referrer,
 			VALIDATION_LIMITS.STRING_MAX_LENGTH
 		),
-		url: sanitizeString(trackData.path, VALIDATION_LIMITS.STRING_MAX_LENGTH),
-		path: sanitizeString(trackData.path, VALIDATION_LIMITS.STRING_MAX_LENGTH),
+		url: sanitizeUrl(trackData.path, VALIDATION_LIMITS.STRING_MAX_LENGTH),
+		path: sanitizeUrl(trackData.path, VALIDATION_LIMITS.STRING_MAX_LENGTH),
 		title: sanitizeString(trackData.title, VALIDATION_LIMITS.STRING_MAX_LENGTH),
 		ip: ctx.geo.anonymizedIP || "",
 		user_agent: "",
@@ -242,7 +243,7 @@ export function insertOutgoingLink(
 				salt
 			),
 			session_id: validateSessionId(linkData.sessionId),
-			href: sanitizeString(linkData.href, VALIDATION_LIMITS.PATH_MAX_LENGTH),
+			href: sanitizeUrl(linkData.href, VALIDATION_LIMITS.PATH_MAX_LENGTH),
 			text: sanitizeString(linkData.text, VALIDATION_LIMITS.TEXT_MAX_LENGTH),
 			properties: linkData.properties
 				? JSON.stringify(linkData.properties)
@@ -293,7 +294,7 @@ export function insertErrorSpans(
 			),
 			session_id: validateSessionId(error.sessionId),
 			timestamp: typeof error.timestamp === "number" ? error.timestamp : now,
-			path: sanitizeString(error.path, VALIDATION_LIMITS.STRING_MAX_LENGTH),
+			path: sanitizeUrl(error.path, VALIDATION_LIMITS.STRING_MAX_LENGTH),
 			message: sanitizeString(
 				error.message,
 				VALIDATION_LIMITS.STRING_MAX_LENGTH
@@ -342,7 +343,7 @@ export function insertIndividualVitals(
 			),
 			session_id: validateSessionId(vital.sessionId),
 			timestamp: typeof vital.timestamp === "number" ? vital.timestamp : now,
-			path: sanitizeString(vital.path, VALIDATION_LIMITS.STRING_MAX_LENGTH),
+			path: sanitizeUrl(vital.path, VALIDATION_LIMITS.STRING_MAX_LENGTH),
 			metric_name: vital.metricName,
 			metric_value: vital.metricValue,
 		}));
@@ -407,7 +408,7 @@ export function insertCustomEvents(
 					)
 				: undefined,
 			path: event.path
-				? sanitizeString(event.path, VALIDATION_LIMITS.STRING_MAX_LENGTH)
+				? sanitizeUrl(event.path, VALIDATION_LIMITS.STRING_MAX_LENGTH)
 				: undefined,
 			properties: event.properties ? JSON.stringify(event.properties) : "{}",
 			anonymous_id: event.anonymous_id

apps/basket/src/routes/identify.test.ts

@@ -4,7 +4,10 @@ vi.mock("@databuddy/db", () => ({
 	db: {},
 	profiles: {},
 	profileAliases: {},
+	profileTraitChanges: {},
 	sql: () => {},
+	and: vi.fn(),
+	eq: vi.fn(),
 }));
 vi.mock("@databuddy/redis/rate-limit", () => ({ ratelimit: vi.fn() }));
 vi.mock("@lib/request-validation", () => ({
@@ -19,6 +22,7 @@ vi.mock("@lib/api-key", () => ({
 vi.mock("evlog/elysia", () => ({ useLogger: () => ({ set: vi.fn() }) }));
 
 import {
+	applyTraits,
 	emailLookupHash,
 	protectPii,
 	revealPii,
@@ -95,6 +99,52 @@ describe("splitTraits", () => {
 	});
 });
 
+describe("applyTraits", () => {
+	test("first identify reports every trait as a change from null", () => {
+		const { changes, traits } = applyTraits({}, { plan: "free", seats: 1 }, []);
+		expect(traits).toEqual({ plan: "free", seats: 1 });
+		expect(changes).toEqual([
+			{ traitKey: "plan", oldValue: null, newValue: "free" },
+			{ traitKey: "seats", oldValue: null, newValue: 1 },
+		]);
+	});
+
+	test("changed value carries old and new, unchanged values are silent", () => {
+		const { changes, traits } = applyTraits(
+			{ plan: "free", seats: 1 },
+			{ plan: "pro", seats: 1 },
+			[]
+		);
+		expect(traits).toEqual({ plan: "pro", seats: 1 });
+		expect(changes).toEqual([
+			{ traitKey: "plan", oldValue: "free", newValue: "pro" },
+		]);
+	});
+
+	test("removed keys drop from the snapshot and report null", () => {
+		const { changes, traits } = applyTraits(
+			{ plan: "pro", beta: true },
+			{},
+			["beta"]
+		);
+		expect(traits).toEqual({ plan: "pro" });
+		expect(changes).toEqual([
+			{ traitKey: "beta", oldValue: true, newValue: null },
+		]);
+	});
+
+	test("removing an absent key changes nothing", () => {
+		const { changes, traits } = applyTraits({ plan: "pro" }, {}, ["missing"]);
+		expect(traits).toEqual({ plan: "pro" });
+		expect(changes).toEqual([]);
+	});
+
+	test("type changes between same-looking values are detected", () => {
+		const { changes } = applyTraits({ seats: "1" }, { seats: 1 }, []);
+		expect(changes).toEqual([{ traitKey: "seats", oldValue: "1", newValue: 1 }]);
+	});
+});
+
 describe("pii protection", () => {
 	const KEY = "test-identity-encryption-key";
 

apps/basket/src/routes/identify.ts

@@ -147,7 +147,7 @@ export const identifyRoute = new Elysia().post(
 				hasAlias: Boolean(anonymousId),
 			});
 
-			await Promise.all([
+			const [update] = await Promise.all([
 				record("upsertProfile", () =>
 					upsertProfile(websiteId, profileId, splitTraits(traits))
 				),
@@ -158,6 +158,10 @@ export const identifyRoute = new Elysia().post(
 					: Promise.resolve(),
 			]);
 
+			if (update && update.changes.length > 0) {
+				log.set({ traitChanges: update.changes.length });
+			}
+
 			return Response.json({ status: "success", type: "identify" });
 		} catch (error) {
 			rethrowOrWrap(error, log);