From b8215a8595e08ddf3aefbe7aee7132767deaaf95 Mon Sep 17 00:00:00 2001 From: mantrakp04 Date: Thu, 2 Jul 2026 11:34:34 -0700 Subject: [PATCH] feat(analytics): public trackEvent/startSpan SDK + custom events/spans ingestion Client SDK (browser): - app.trackEvent(type, data?, {parentIds?}) and app.startSpan(type, {data?, parentIds?, startedAtMs?}) on StackClientApp - Span handles: setData (merge + re-write), end (idempotent), trackEvent/startSpan (inherit full chain), ref() for cross-tier parenting - setGlobalSpan/unsetGlobalSpan (ambient parents for spans AND events, auto-unset on end), app.flush() - Per-item promises settle on batch ack; pre-caught (no unhandled rejections); nothing throws synchronously - Spans are versioned upserts: written open on start, re-written on setData/end with per-handle monotonic updated_at_ms - Batch-poisoning protection: full item validation at enqueue (name regex, 16KB data cap, uuid parents, depth<=10) - clearBuffer rejects pending + inert-ifies live spans (sign-out privacy, pairs with segment-id rotation) Server SDK: - serverApp.trackEvent/startSpan with explicit {userId}; same-tick coalescing per userId (one POST per sync burst) - sendAnalyticsEventBatchAsServer on HexclaveServerInterface (secret server key -> analytics base URL) Backend: - events batch route accepts custom event types, events[*].parent_span_ids, spans[], and server/admin auth with body user_id (verified in tenancy) - buildCustomSpanRows: cs- prefixed ids, system ancestry (rti-/sri-/srsi-, same gating as events) prepended to client chain, version clamped to now+5min - custom spans insert off the response path (waitUntil), quota debits events+spans - stripLoneSurrogates moved to lib/clickhouse for reuse Tests: 8 new spans.test.ts cases, 9 new event-tracker.test.ts cases, 17 new e2e cases (e2e not executed: dev stack down; 2 pre-existing snapshots updated for intentional schema-message changes) --- .../latest/analytics/events/batch/route.tsx | 223 ++++++-- apps/backend/src/lib/clickhouse.tsx | 22 + apps/backend/src/lib/spans.test.ts | 120 +++- apps/backend/src/lib/spans.tsx | 84 ++- .../api/v1/analytics-events-batch.test.ts | 520 ++++++++++++++++- .../shared/src/interface/client-interface.ts | 2 +- .../shared/src/interface/server-interface.ts | 33 ++ packages/template/src/index.ts | 4 + .../apps/implementations/client-app-impl.ts | 46 +- .../implementations/event-tracker.test.ts | 314 +++++++++++ .../apps/implementations/event-tracker.ts | 528 +++++++++++++++++- .../apps/implementations/server-app-impl.ts | 292 ++++++++++ .../apps/interfaces/client-app.ts | 33 ++ .../apps/interfaces/server-app.ts | 15 + 14 files changed, 2141 insertions(+), 95 deletions(-) diff --git a/apps/backend/src/app/api/latest/analytics/events/batch/route.tsx b/apps/backend/src/app/api/latest/analytics/events/batch/route.tsx index a9e063003..a94c1b07e 100644 --- a/apps/backend/src/app/api/latest/analytics/events/batch/route.tsx +++ b/apps/backend/src/app/api/latest/analytics/events/batch/route.tsx @@ -1,42 +1,35 @@ -import { getClickhouseAdminClient } from "@/lib/clickhouse"; +import { getClickhouseAdminClient, stripLoneSurrogates } from "@/lib/clickhouse"; import { arePlanLimitsEnforced, getBillingTeamId } from "@/lib/plan-entitlements"; import { findRecentSessionReplay } from "@/lib/session-replays"; -import { buildEventSpanFields } from "@/lib/spans"; +import { buildCustomSpanRows, buildEventSpanFields, insertSpans, toSpanId, SPAN_ID_PREFIXES } from "@/lib/spans"; +import { runAsynchronouslyAndWaitUntil } from "@/utils/background-tasks"; import { getHexclaveServerApp } from "@/hexclave"; import { getPrismaClientForTenancy } from "@/prisma-client"; import { createSmartRouteHandler } from "@/route-handlers/smart-route-handler"; import { KnownErrors } from "@hexclave/shared"; import { ITEM_IDS } from "@hexclave/shared/dist/plans"; import { adaptSchema, clientOrHigherAuthTypeSchema, yupArray, yupMixed, yupNumber, yupObject, yupString } from "@hexclave/shared/dist/schema-fields"; -import { StatusError } from "@hexclave/shared/dist/utils/errors"; +import { captureError, StatusError } from "@hexclave/shared/dist/utils/errors"; import * as zlib from "node:zlib"; const UUID_RE = /^[0-9a-f]{8}-[0-9a-f]{4}-[1-8][0-9a-f]{3}-[089ab][0-9a-f]{3}-[0-9a-f]{12}$/i; +// Custom (user-defined) event/span type names: must not start with `$` (reserved +// for system types), start with a letter, and stay within 64 chars. Keep in sync +// with the SDK-side validation in packages/template's event-tracker. +const CUSTOM_TELEMETRY_NAME_RE = /^[a-zA-Z][a-zA-Z0-9_.:-]{0,63}$/; +const SYSTEM_EVENT_TYPES = ["$page-view", "$click"] as const; + const MAX_EVENTS = 500; +const MAX_SPANS = 500; +const MAX_PARENT_CHAIN = 10; +const MAX_ITEM_DATA_BYTES = 16_000; const MAX_COMPRESSED_BYTES = 1 * 1024 * 1024; const MAX_DECOMPRESSED_BYTES = 8 * 1024 * 1024; -// Lone surrogates (\uD800-\uDFFF not part of a valid pair) are technically -// representable in JS strings but rejected by ClickHouse's JSON parser. -// The client-side event tracker can produce these when .substring() truncates -// text in the middle of a surrogate pair (e.g. emoji characters). -// eslint-disable-next-line no-control-regex -const LONE_SURROGATE_RE = /[\uD800-\uDBFF](?![\uDC00-\uDFFF])|(? [k, stripLoneSurrogates(v)]) - ); - } - return value; +function isPlainObjectWithinLimit(value: unknown): boolean { + if (value === null || typeof value !== "object" || Array.isArray(value)) return false; + return JSON.stringify(value).length <= MAX_ITEM_DATA_BYTES; } // Bodies sent as application/octet-stream are gzipped JSON. The encoding is @@ -70,8 +63,8 @@ function maybeDecodeBinaryBody(value: unknown): unknown { export const POST = createSmartRouteHandler({ metadata: { - summary: "Upload analytics event batch", - description: "Uploads a batch of auto-captured analytics events ($page-view, $click).", + summary: "Upload analytics telemetry batch", + description: "Uploads a batch of analytics telemetry: auto-captured events ($page-view, $click), custom events, and custom spans.", tags: ["Analytics Events"], hidden: true, }, @@ -83,17 +76,61 @@ export const POST = createSmartRouteHandler({ refreshTokenId: adaptSchema, }).defined(), body: yupObject({ - session_replay_segment_id: yupString().defined().matches(UUID_RE, "Invalid session_replay_segment_id"), + // Required for client auth (enforced in the handler — browser batches are + // always tied to a per-tab segment); optional for server/admin auth. + session_replay_segment_id: yupString().optional().matches(UUID_RE, "Invalid session_replay_segment_id"), batch_id: yupString().defined().matches(UUID_RE, "Invalid batch_id"), sent_at_ms: yupNumber().defined().integer().min(0), + // Server/admin auth only (enforced in the handler): attributes the batch + // to a user when there is no session access token to derive one from. + user_id: yupString().optional().matches(UUID_RE, "Invalid user_id"), events: yupArray( yupObject({ - event_type: yupString().defined().oneOf(["$page-view", "$click"]), + event_type: yupString().defined().test( + "event-type", + `event_type must be one of ${SYSTEM_EVENT_TYPES.join(", ")} or a custom name matching ${CUSTOM_TELEMETRY_NAME_RE}`, + // yup skips tests for undefined values, so `value` is always set here. + (value) => (SYSTEM_EVENT_TYPES as readonly string[]).includes(value) || CUSTOM_TELEMETRY_NAME_RE.test(value), + ), event_at_ms: yupNumber().defined().integer().min(0), data: yupMixed().defined(), - }).defined(), - ).defined().min(1).max(MAX_EVENTS), - }).defined().transform((_value, originalValue) => maybeDecodeBinaryBody(originalValue)), + // Custom ancestor chain, root-first, raw span uuids. System ancestry + // (refresh-token/replay/segment) is composed server-side on top. + parent_span_ids: yupArray(yupString().defined().matches(UUID_RE, "Invalid parent span id")).optional().max(MAX_PARENT_CHAIN), + }).defined().test( + "custom-event-data", + `Custom event data must be a JSON object of at most ${MAX_ITEM_DATA_BYTES} serialized bytes`, + // System event data stays permissive for backward compatibility with + // deployed trackers; the object/size cap applies to custom types only. + (event) => (SYSTEM_EVENT_TYPES as readonly string[]).includes(event.event_type) || isPlainObjectWithinLimit(event.data), + ), + ).optional().max(MAX_EVENTS), + spans: yupArray( + yupObject({ + span_id: yupString().defined().matches(UUID_RE, "Invalid span_id"), + // Custom names only — `$…` span types are reserved for system spans + // and can never be written through this endpoint. + span_type: yupString().defined().matches(CUSTOM_TELEMETRY_NAME_RE, "Invalid span_type"), + started_at_ms: yupNumber().defined().integer().min(0), + ended_at_ms: yupNumber().nullable().defined().integer().min(0), + parent_span_ids: yupArray(yupString().defined().matches(UUID_RE, "Invalid parent span id")).defined().max(MAX_PARENT_CHAIN), + data: yupMixed().defined().test( + "span-data", + `Span data must be a JSON object of at most ${MAX_ITEM_DATA_BYTES} serialized bytes`, + (value) => isPlainObjectWithinLimit(value), + ), + updated_at_ms: yupNumber().defined().integer().min(0), + }).defined().test( + "span-interval", + "ended_at_ms must be greater than or equal to started_at_ms", + (span) => span.ended_at_ms == null || span.ended_at_ms >= span.started_at_ms, + ), + ).optional().max(MAX_SPANS), + }).defined().test( + "non-empty-batch", + "A batch must contain at least one event or span", + (body) => (body.events?.length ?? 0) + (body.spans?.length ?? 0) >= 1, + ).transform((_value, originalValue) => maybeDecodeBinaryBody(originalValue)), }), response: yupObject({ statusCode: yupNumber().oneOf([200]).defined(), @@ -106,45 +143,84 @@ export const POST = createSmartRouteHandler({ if (!auth.tenancy.config.apps.installed["analytics"]?.enabled) { throw new KnownErrors.AnalyticsNotEnabled(); } - if (!auth.user) { - throw new KnownErrors.UserAuthenticationRequired(); - } - if (!auth.refreshTokenId) { - throw new StatusError(StatusError.BadRequest, "A refresh token is required for analytics events"); + + const events = body.events ?? []; + const spans = body.spans ?? []; + const tenancyId = auth.tenancy.id; + const prisma = await getPrismaClientForTenancy(auth.tenancy); + + // Client auth is the browser tracker: identity comes from the session (user + + // refresh token, always present) and batches are tied to a per-tab segment. + // Server/admin auth has no session to derive from, so the caller attributes + // the batch explicitly via body.user_id (or not at all — project-level rows). + let userId: string | null; + let refreshTokenId: string | null; + if (auth.type === "client") { + if (!auth.user) { + throw new KnownErrors.UserAuthenticationRequired(); + } + if (!auth.refreshTokenId) { + throw new StatusError(StatusError.BadRequest, "A refresh token is required for analytics events"); + } + if (body.user_id != null) { + throw new StatusError(StatusError.BadRequest, "user_id must not be set with client auth; it is derived from the session"); + } + if (body.session_replay_segment_id == null) { + throw new StatusError(StatusError.BadRequest, "session_replay_segment_id is required for analytics batches with client auth"); + } + userId = auth.user.id; + refreshTokenId = auth.refreshTokenId; + } else { + if (body.user_id != null) { + const user = await prisma.projectUser.findUnique({ + where: { + tenancyId_projectUserId: { + tenancyId, + projectUserId: body.user_id, + }, + }, + }); + if (!user) { + throw new StatusError(StatusError.BadRequest, "user_id does not correspond to a user on this project/branch"); + } + } + userId = body.user_id ?? auth.user?.id ?? null; + refreshTokenId = auth.refreshTokenId ?? null; } const projectId = auth.tenancy.project.id; const branchId = auth.tenancy.branchId; - const userId = auth.user.id; - const refreshTokenId = auth.refreshTokenId; - const tenancyId = auth.tenancy.id; const app = getHexclaveServerApp(); const billingTeamId = getBillingTeamId(auth.tenancy.project); if (billingTeamId != null && arePlanLimitsEnforced()) { + const totalItems = events.length + spans.length; const eventsItem = await app.getItem({ itemId: ITEM_IDS.analyticsEvents, teamId: billingTeamId }); - const isDebited = await eventsItem.tryDecreaseQuantity(body.events.length); + const isDebited = await eventsItem.tryDecreaseQuantity(totalItems); if (!isDebited) { - throw new KnownErrors.ItemQuantityInsufficientAmount(ITEM_IDS.analyticsEvents, billingTeamId, body.events.length); + throw new KnownErrors.ItemQuantityInsufficientAmount(ITEM_IDS.analyticsEvents, billingTeamId, totalItems); } } - const prisma = await getPrismaClientForTenancy(auth.tenancy); - const recentSession = await findRecentSessionReplay(prisma, { tenancyId, refreshTokenId }); + const recentSession = refreshTokenId == null ? null : await findRecentSessionReplay(prisma, { tenancyId, refreshTokenId }); + const sessionReplayId = recentSession?.id ?? null; + const sessionReplaySegmentId = body.session_replay_segment_id ?? null; const clickhouseClient = getClickhouseAdminClient(); // Point each event at its ancestor spans (root-first: refresh-token, replay, // then the per-tab span when a replay exists). The per-tab id itself already - // lives in the session_replay_segment_id column — see lib/spans.tsx. + // lives in the session_replay_segment_id column — see lib/spans.tsx. The + // client-supplied custom chain (raw uuids) is appended after the system + // ancestry, each id prefixed cs-. const eventSpanFields = buildEventSpanFields({ - sessionReplayId: recentSession?.id ?? null, - sessionReplaySegmentId: body.session_replay_segment_id, + sessionReplayId, + sessionReplaySegmentId, refreshTokenId, }); - const rows = body.events.map((event) => ({ + const rows = events.map((event) => ({ event_type: event.event_type, event_at: new Date(event.event_at_ms), data: stripLoneSurrogates(event.data), @@ -153,25 +229,54 @@ export const POST = createSmartRouteHandler({ user_id: userId, team_id: null, refresh_token_id: refreshTokenId, - session_replay_id: recentSession?.id ?? null, - session_replay_segment_id: body.session_replay_segment_id, - ...eventSpanFields, + session_replay_id: sessionReplayId, + session_replay_segment_id: sessionReplaySegmentId, + parent_span_ids: [ + ...eventSpanFields.parent_span_ids, + ...(event.parent_span_ids ?? []).map((id) => toSpanId(SPAN_ID_PREFIXES.custom, id)), + ], })); - await clickhouseClient.insert({ - table: "analytics_internal.events", - values: rows, - format: "JSONEachRow", - clickhouse_settings: { - date_time_input_format: "best_effort", - async_insert: 1, - }, - }); + if (rows.length > 0) { + await clickhouseClient.insert({ + table: "analytics_internal.events", + values: rows, + format: "JSONEachRow", + clickhouse_settings: { + date_time_input_format: "best_effort", + async_insert: 1, + }, + }); + } + + // Custom spans are versioned upserts into analytics_internal.spans. Written + // off the response path (same pattern as the replay batch route) so a slow or + // unavailable ClickHouse never delays the upload; the events insert above + // stays on-path because the response reports what was accepted. + if (spans.length > 0) { + const spanRows = buildCustomSpanRows({ + spans, + projectId, + branchId, + userId, + refreshTokenId, + sessionReplayId, + sessionReplaySegmentId, + serverNowMs: Date.now(), + }); + runAsynchronouslyAndWaitUntil(async () => { + try { + await insertSpans(clickhouseClient, spanRows); + } catch (error) { + captureError("analytics-custom-spans-insert", error); + } + }); + } return { statusCode: 200, bodyType: "json", - body: { inserted: body.events.length }, + body: { inserted: events.length + spans.length }, }; }, }); diff --git a/apps/backend/src/lib/clickhouse.tsx b/apps/backend/src/lib/clickhouse.tsx index 26344e880..0e52ef58d 100644 --- a/apps/backend/src/lib/clickhouse.tsx +++ b/apps/backend/src/lib/clickhouse.tsx @@ -7,6 +7,28 @@ import { HexclaveAssertionError } from "@hexclave/shared/dist/utils/errors"; // dependency on the @clickhouse/client package. export type { ClickHouseClient } from "@clickhouse/client"; +// Lone surrogates (\uD800-\uDFFF not part of a valid pair) are technically +// representable in JS strings but rejected by ClickHouse's JSON parser. +// The client-side event tracker can produce these when .substring() truncates +// text in the middle of a surrogate pair (e.g. emoji characters). +// eslint-disable-next-line no-control-regex +const LONE_SURROGATE_RE = /[\uD800-\uDBFF](?![\uDC00-\uDFFF])|(? [k, stripLoneSurrogates(v)]) + ); + } + return value; +} + function getAdminAuth() { return { username: getEnvVariable("STACK_CLICKHOUSE_ADMIN_USER", "stackframe"), diff --git a/apps/backend/src/lib/spans.test.ts b/apps/backend/src/lib/spans.test.ts index a61fa5931..cf0e5893b 100644 --- a/apps/backend/src/lib/spans.test.ts +++ b/apps/backend/src/lib/spans.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it, vi } from "vitest"; -import { SPAN_ID_PREFIXES, buildEventSpanFields, insertSessionReplaySpans, toSpanId } from "./spans"; +import { SPAN_ID_PREFIXES, buildCustomSpanRows, buildEventSpanFields, insertSessionReplaySpans, toSpanId } from "./spans"; describe("toSpanId", () => { it("prefixes raw ids without touching the raw value", () => { @@ -103,3 +103,121 @@ describe("insertSessionReplaySpans", () => { expect(client.insert).not.toHaveBeenCalled(); }); }); + +describe("buildCustomSpanRows", () => { + const NOW_MS = new Date("2026-01-01T00:10:00.000Z").getTime(); + const baseOpts = { + projectId: "p1", + branchId: "b1", + userId: "user1", + refreshTokenId: "rt1", + sessionReplayId: "replay1", + sessionReplaySegmentId: "seg1", + serverNowMs: NOW_MS, + }; + const baseSpan = { + span_id: "0f000000-0000-4000-8000-000000000001", + span_type: "checkout-flow", + started_at_ms: NOW_MS - 60_000, + ended_at_ms: null, + parent_span_ids: [] as string[], + data: {}, + updated_at_ms: NOW_MS - 60_000, + }; + + it("prefixes the span id and every client parent with cs- and prepends the full system ancestry root-first", () => { + const rows = buildCustomSpanRows({ + ...baseOpts, + spans: [{ + ...baseSpan, + parent_span_ids: ["0f000000-0000-4000-8000-00000000aaaa", "0f000000-0000-4000-8000-00000000bbbb"], + }], + }); + expect(rows).toHaveLength(1); + expect(rows[0].id).toBe(`cs-${baseSpan.span_id}`); + expect(rows[0].parent_span_ids).toEqual([ + "rti-rt1", + "sri-replay1", + "srsi-seg1", + "cs-0f000000-0000-4000-8000-00000000aaaa", + "cs-0f000000-0000-4000-8000-00000000bbbb", + ]); + }); + + it("degrades the system ancestry with the same gating as event rows (refresh-token only, then none)", () => { + const refreshTokenOnly = buildCustomSpanRows({ + ...baseOpts, + sessionReplayId: null, + sessionReplaySegmentId: null, + spans: [baseSpan], + }); + expect(refreshTokenOnly[0].parent_span_ids).toEqual(["rti-rt1"]); + + const serverAuth = buildCustomSpanRows({ + ...baseOpts, + userId: null, + refreshTokenId: null, + sessionReplayId: null, + sessionReplaySegmentId: null, + spans: [baseSpan], + }); + expect(serverAuth[0].parent_span_ids).toEqual([]); + expect(serverAuth[0].user_id).toBeNull(); + expect(serverAuth[0].refresh_token_id).toBeNull(); + }); + + it("keeps open intervals open and fills identity columns raw (unprefixed)", () => { + const rows = buildCustomSpanRows({ ...baseOpts, spans: [baseSpan] }); + expect(rows[0]).toMatchObject({ + span_type: "checkout-flow", + span_ended_at: null, + project_id: "p1", + branch_id: "b1", + user_id: "user1", + team_id: null, + refresh_token_id: "rt1", + session_replay_id: "replay1", + session_replay_segment_id: "seg1", + }); + expect(rows[0].span_started_at).toEqual(new Date(baseSpan.started_at_ms)); + }); + + it("uses the client updated_at_ms as the version, clamped to [1, now + 5min]", () => { + const rows = buildCustomSpanRows({ + ...baseOpts, + spans: [ + { ...baseSpan, updated_at_ms: NOW_MS - 1_000 }, + { ...baseSpan, span_id: "0f000000-0000-4000-8000-000000000002", updated_at_ms: 0 }, + { ...baseSpan, span_id: "0f000000-0000-4000-8000-000000000003", updated_at_ms: NOW_MS + 60 * 60 * 1000 }, + ], + }); + expect(rows[0].version).toBe(NOW_MS - 1_000); + expect(rows[1].version).toBe(1); + expect(rows[2].version).toBe(NOW_MS + 5 * 60 * 1000); + }); + + it("serializes data to a JSON string and strips lone surrogates (ClickHouse rejects them)", () => { + const rows = buildCustomSpanRows({ + ...baseOpts, + spans: [{ ...baseSpan, data: { label: "truncated \uD83D", count: 3 } }], + }); + expect(rows[0].data).toBe(JSON.stringify({ label: "truncated �", count: 3 })); + }); + + it("closes the interval when ended_at_ms is set", () => { + const endedAtMs = NOW_MS - 30_000; + const rows = buildCustomSpanRows({ + ...baseOpts, + spans: [{ ...baseSpan, ended_at_ms: endedAtMs, updated_at_ms: endedAtMs }], + }); + expect(rows[0].span_ended_at).toEqual(new Date(endedAtMs)); + expect(rows[0].version).toBe(endedAtMs); + }); + + it("refuses $-prefixed span types even if a caller bypasses the route schema", () => { + expect(() => buildCustomSpanRows({ + ...baseOpts, + spans: [{ ...baseSpan, span_type: "$session-replay" }], + })).toThrowError(/must not start with "\$"/); + }); +}); diff --git a/apps/backend/src/lib/spans.tsx b/apps/backend/src/lib/spans.tsx index 639213e5a..b2ac8009f 100644 --- a/apps/backend/src/lib/spans.tsx +++ b/apps/backend/src/lib/spans.tsx @@ -1,4 +1,5 @@ -import type { ClickHouseClient } from "./clickhouse"; +import { HexclaveAssertionError } from "@hexclave/shared/dist/utils/errors"; +import { stripLoneSurrogates, type ClickHouseClient } from "./clickhouse"; /** * Spans are telemetry facts (time intervals) — the sibling of events. They are @@ -24,6 +25,10 @@ export const SPAN_ID_PREFIXES = { team: "ti-", project: "pi-", branch: "bi-", + // User-defined spans created via the SDK's startSpan(). Applied server-side to + // the client-generated raw uuid and to every id inside a client-supplied parent + // chain — clients only ever transmit raw custom uuids, never prefixed ids. + custom: "cs-", } as const; export const SPAN_TYPES = { @@ -102,6 +107,83 @@ export async function insertSpans(client: ClickHouseClient, rows: SpanInsertRow[ }); } +/** + * One custom span as it arrives on the wire from the SDK (inside the analytics + * events batch). Ids are raw uuids; `parent_span_ids` is the client's CUSTOM + * ancestor chain only (root-first) — system ancestry is composed server-side. + */ +export type CustomSpanWireItem = { + span_id: string, + span_type: string, + started_at_ms: number, + ended_at_ms: number | null, + parent_span_ids: string[], + data: unknown, + updated_at_ms: number, +}; + +// How far into the future a client-supplied `updated_at_ms` may run before we +// clamp it. A skewed clock only corrupts ordering among that user's own span +// re-writes; the clamp just bounds how long a bogus future version could mask +// legitimate later updates. +const CUSTOM_SPAN_VERSION_MAX_FUTURE_MS = 5 * 60 * 1000; + +/** + * Builds `analytics_internal.spans` rows for SDK-created custom spans. Each + * row's `parent_span_ids` is the server-known system ancestry (same gating as + * event rows — see `buildEventSpanFields`) followed by the client's custom + * chain, every custom id prefixed `cs-`. The version is the client's + * `updated_at_ms` — per-span monotonic by SDK construction, so the row carrying + * the latest update wins in the ReplacingMergeTree regardless of insert order + * (an end row can never be shadowed by a late-arriving open row). The replay + * spans' end-time-as-version scheme is unusable here: two data re-writes while + * the span is still open would collide at the same version. + */ +export function buildCustomSpanRows(opts: { + spans: CustomSpanWireItem[], + projectId: string, + branchId: string, + userId: string | null, + refreshTokenId: string | null, + sessionReplayId: string | null, + sessionReplaySegmentId: string | null, + serverNowMs: number, +}): SpanInsertRow[] { + const systemAncestry = buildEventSpanFields({ + sessionReplayId: opts.sessionReplayId, + sessionReplaySegmentId: opts.sessionReplaySegmentId, + refreshTokenId: opts.refreshTokenId, + }).parent_span_ids; + + return opts.spans.map((span) => { + // The route schema is the primary gate; this backstop keeps the invariant + // even for future callers: `$…` span types are reserved for system spans + // and must never be writable through the custom-span path. + if (span.span_type.startsWith("$")) { + throw new HexclaveAssertionError(`Custom span types must not start with "$". Received: ${JSON.stringify(span.span_type)}`); + } + return { + id: toSpanId(SPAN_ID_PREFIXES.custom, span.span_id), + span_type: span.span_type, + span_started_at: new Date(span.started_at_ms), + span_ended_at: span.ended_at_ms == null ? null : new Date(span.ended_at_ms), + parent_span_ids: [ + ...systemAncestry, + ...span.parent_span_ids.map((id) => toSpanId(SPAN_ID_PREFIXES.custom, id)), + ], + data: JSON.stringify(stripLoneSurrogates(span.data)), + project_id: opts.projectId, + branch_id: opts.branchId, + user_id: opts.userId, + team_id: null, + refresh_token_id: opts.refreshTokenId, + session_replay_id: opts.sessionReplayId, + session_replay_segment_id: opts.sessionReplaySegmentId, + version: Math.min(Math.max(span.updated_at_ms, 1), opts.serverNowMs + CUSTOM_SPAN_VERSION_MAX_FUTURE_MS), + }; + }); +} + /** * Emits the two spans that describe a session replay from the replay batch route: * the replay-level `$session-replay` span and the per-tab `$session-replay-segment` diff --git a/apps/e2e/tests/backend/endpoints/api/v1/analytics-events-batch.test.ts b/apps/e2e/tests/backend/endpoints/api/v1/analytics-events-batch.test.ts index bc6c9eb88..ccf8f552a 100644 --- a/apps/e2e/tests/backend/endpoints/api/v1/analytics-events-batch.test.ts +++ b/apps/e2e/tests/backend/endpoints/api/v1/analytics-events-batch.test.ts @@ -343,12 +343,12 @@ it("rejects empty events array", async ({ expect }) => { "details": { "message": deindent\` Request validation failed on POST /api/v1/analytics/events/batch: - - body.events field must have at least 1 items + - A batch must contain at least one event or span \`, }, "error": deindent\` Request validation failed on POST /api/v1/analytics/events/batch: - - body.events field must have at least 1 items + - A batch must contain at least one event or span \`, }, "headers": Headers { @@ -505,12 +505,12 @@ it("rejects invalid event_type", async ({ expect }) => { "details": { "message": deindent\` Request validation failed on POST /api/v1/analytics/events/batch: - - body.events[0].event_type must be one of the following values: $page-view, $click + - event_type must be one of $page-view, $click or a custom name matching /^[a-zA-Z][a-zA-Z0-9_.:-]{0,63}$/ \`, }, "error": deindent\` Request validation failed on POST /api/v1/analytics/events/batch: - - body.events[0].event_type must be one of the following values: $page-view, $click + - event_type must be one of $page-view, $click or a custom name matching /^[a-zA-Z][a-zA-Z0-9_.:-]{0,63}$/ \`, }, "headers": Headers { @@ -770,3 +770,515 @@ it("team plan starts with correct analytics event allocation", async ({ expect } const quantity = await getItemQuantity(ownerTeamId, ITEM_IDS.analyticsEvents); expect(quantity).toBe(PLAN_LIMITS.team.analyticsEvents); }); + +// ============================================================================ +// Custom events & custom spans +// ============================================================================ + +async function setupAnalyticsProject() { + await Project.createAndSwitch({ config: { magic_link_enabled: true } }); + await Project.updateConfig({ apps: { installed: { analytics: { enabled: true } } } }); +} + +async function uploadTelemetryBatch( + body: { + session_replay_segment_id?: string, + batch_id?: string, + sent_at_ms?: number, + user_id?: string, + events?: unknown[], + spans?: unknown[], + }, + options?: { accessType?: "client" | "server" }, +) { + return await niceBackendFetch("/api/v1/analytics/events/batch", { + method: "POST", + accessType: options?.accessType ?? "client", + body: { + batch_id: randomUUID(), + sent_at_ms: Date.now(), + ...body, + }, + }); +} + +function makeCustomSpan(overrides?: Record) { + const now = Date.now(); + return { + span_id: randomUUID(), + span_type: "checkout-flow", + started_at_ms: now - 1000, + ended_at_ms: null, + parent_span_ids: [], + data: {}, + updated_at_ms: now, + ...overrides, + }; +} + +// Retry query because both the events insert (async_insert) and the spans +// insert (written off the response path via waitUntil) land with a delay. +async function queryAnalyticsUntil( + body: { query: string, params?: Record }, + isDone: (res: { status: number, body: any }) => boolean, + attempts = 20, +) { + let queryRes; + for (let attempt = 0; attempt < attempts; attempt++) { + await wait(500); + queryRes = await niceBackendFetch("/api/v1/internal/analytics/query", { + method: "POST", + accessType: "admin", + body, + }); + if (queryRes.status === 200 && isDone(queryRes)) { + break; + } + } + return queryRes; +} + +it("accepts custom events and stamps system ancestry on them", async ({ expect }) => { + await setupAnalyticsProject(); + await Auth.Otp.signIn(); + + const sessionReplaySegmentId = randomUUID(); + const now = Date.now(); + const res = await uploadTelemetryBatch({ + session_replay_segment_id: sessionReplaySegmentId, + events: [{ event_type: "checkout_completed", event_at_ms: now - 100, data: { cart_size: 3 } }], + }); + expect(res).toMatchInlineSnapshot(` + NiceResponse { + "status": 200, + "body": { "inserted": 1 }, + "headers": Headers {