mirror of
https://github.com/stack-auth/stack.git
synced 2026-07-20 21:29:36 +08:00
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)
This commit is contained in:
mantrakp04
parent
ddec2a4ac4
commit
b8215a8595
@ -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])|(?<![\uD800-\uDBFF])[\uDC00-\uDFFF]/g;
|
||||
|
||||
function stripLoneSurrogates(value: unknown): unknown {
|
||||
if (typeof value === "string") {
|
||||
return value.replace(LONE_SURROGATE_RE, "\uFFFD");
|
||||
}
|
||||
if (Array.isArray(value)) {
|
||||
return value.map(stripLoneSurrogates);
|
||||
}
|
||||
if (value !== null && typeof value === "object") {
|
||||
return Object.fromEntries(
|
||||
Object.entries(value).map(([k, v]) => [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 },
|
||||
};
|
||||
},
|
||||
});
|
||||
|
||||
@ -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])|(?<![\uD800-\uDBFF])[\uDC00-\uDFFF]/g;
|
||||
|
||||
export function stripLoneSurrogates(value: unknown): unknown {
|
||||
if (typeof value === "string") {
|
||||
return value.replace(LONE_SURROGATE_RE, "<22>");
|
||||
}
|
||||
if (Array.isArray(value)) {
|
||||
return value.map(stripLoneSurrogates);
|
||||
}
|
||||
if (value !== null && typeof value === "object") {
|
||||
return Object.fromEntries(
|
||||
Object.entries(value).map(([k, v]) => [k, stripLoneSurrogates(v)])
|
||||
);
|
||||
}
|
||||
return value;
|
||||
}
|
||||
|
||||
function getAdminAuth() {
|
||||
return {
|
||||
username: getEnvVariable("STACK_CLICKHOUSE_ADMIN_USER", "stackframe"),
|
||||
|
||||
@ -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 <20>", 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 "\$"/);
|
||||
});
|
||||
});
|
||||
|
||||
@ -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`
|
||||
|
||||
@ -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<string, unknown>) {
|
||||
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<string, string> },
|
||||
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 { <some fields may have been hidden> },
|
||||
}
|
||||
`);
|
||||
|
||||
const queryRes = await queryAnalyticsUntil({
|
||||
query: "SELECT event_type, parent_span_ids FROM events WHERE session_replay_segment_id = {segId:String}",
|
||||
params: { segId: sessionReplaySegmentId },
|
||||
}, (r) => r.body?.result?.length === 1);
|
||||
|
||||
expect(queryRes?.status).toBe(200);
|
||||
const row = (queryRes?.body as any).result[0];
|
||||
expect(row.event_type).toBe("checkout_completed");
|
||||
// No active session replay, so the only system ancestor is the refresh-token span.
|
||||
expect(row.parent_span_ids).toHaveLength(1);
|
||||
expect(row.parent_span_ids[0]).toMatch(/^rti-/);
|
||||
});
|
||||
|
||||
it("appends the client-supplied custom parent chain after system ancestry on events", async ({ expect }) => {
|
||||
await setupAnalyticsProject();
|
||||
await Auth.Otp.signIn();
|
||||
|
||||
const sessionReplaySegmentId = randomUUID();
|
||||
const parentSpanId = randomUUID();
|
||||
const now = Date.now();
|
||||
const res = await uploadTelemetryBatch({
|
||||
session_replay_segment_id: sessionReplaySegmentId,
|
||||
events: [{
|
||||
event_type: "checkout_step",
|
||||
event_at_ms: now - 100,
|
||||
data: { step: 1 },
|
||||
parent_span_ids: [parentSpanId],
|
||||
}],
|
||||
});
|
||||
expect(res.status).toBe(200);
|
||||
expect(res.body.inserted).toBe(1);
|
||||
|
||||
const queryRes = await queryAnalyticsUntil({
|
||||
query: "SELECT parent_span_ids FROM events WHERE session_replay_segment_id = {segId:String}",
|
||||
params: { segId: sessionReplaySegmentId },
|
||||
}, (r) => r.body?.result?.length === 1);
|
||||
|
||||
expect(queryRes?.status).toBe(200);
|
||||
const row = (queryRes?.body as any).result[0];
|
||||
// Root-first: system ancestry (refresh-token) first, then the cs-prefixed custom chain.
|
||||
expect(row.parent_span_ids).toHaveLength(2);
|
||||
expect(row.parent_span_ids[0]).toMatch(/^rti-/);
|
||||
expect(row.parent_span_ids[1]).toBe(`cs-${parentSpanId}`);
|
||||
});
|
||||
|
||||
it("rejects unknown $-prefixed event types", async ({ expect }) => {
|
||||
await setupAnalyticsProject();
|
||||
await Auth.Otp.signIn();
|
||||
|
||||
const res = await uploadTelemetryBatch({
|
||||
session_replay_segment_id: randomUUID(),
|
||||
events: [{ event_type: "$custom-fake", event_at_ms: Date.now(), data: {} }],
|
||||
});
|
||||
|
||||
expect(res.status).toBe(400);
|
||||
expect(res.body?.code).toBe("SCHEMA_ERROR");
|
||||
expect(res.body?.error).toContain("event_type must be one of");
|
||||
});
|
||||
|
||||
it("rejects custom event types that do not start with a letter", async ({ expect }) => {
|
||||
await setupAnalyticsProject();
|
||||
await Auth.Otp.signIn();
|
||||
|
||||
const res = await uploadTelemetryBatch({
|
||||
session_replay_segment_id: randomUUID(),
|
||||
events: [{ event_type: "1bad", event_at_ms: Date.now(), data: {} }],
|
||||
});
|
||||
|
||||
expect(res.status).toBe(400);
|
||||
expect(res.body?.code).toBe("SCHEMA_ERROR");
|
||||
expect(res.body?.error).toContain("event_type must be one of");
|
||||
});
|
||||
|
||||
it("rejects custom event types longer than 64 characters", async ({ expect }) => {
|
||||
await setupAnalyticsProject();
|
||||
await Auth.Otp.signIn();
|
||||
|
||||
const res = await uploadTelemetryBatch({
|
||||
session_replay_segment_id: randomUUID(),
|
||||
events: [{ event_type: "a".repeat(65), event_at_ms: Date.now(), data: {} }],
|
||||
});
|
||||
|
||||
expect(res.status).toBe(400);
|
||||
expect(res.body?.code).toBe("SCHEMA_ERROR");
|
||||
expect(res.body?.error).toContain("event_type must be one of");
|
||||
});
|
||||
|
||||
it("rejects custom event data that is not a plain object", async ({ expect }) => {
|
||||
await setupAnalyticsProject();
|
||||
await Auth.Otp.signIn();
|
||||
|
||||
const res = await uploadTelemetryBatch({
|
||||
session_replay_segment_id: randomUUID(),
|
||||
events: [{ event_type: "checkout_completed", event_at_ms: Date.now(), data: [1, 2, 3] }],
|
||||
});
|
||||
|
||||
expect(res.status).toBe(400);
|
||||
expect(res.body?.code).toBe("SCHEMA_ERROR");
|
||||
expect(res.body?.error).toContain("Custom event data must be a JSON object");
|
||||
});
|
||||
|
||||
it("rejects custom event data larger than the serialized size cap", async ({ expect }) => {
|
||||
await setupAnalyticsProject();
|
||||
await Auth.Otp.signIn();
|
||||
|
||||
const res = await uploadTelemetryBatch({
|
||||
session_replay_segment_id: randomUUID(),
|
||||
events: [{ event_type: "checkout_completed", event_at_ms: Date.now(), data: { pad: "x".repeat(16_001) } }],
|
||||
});
|
||||
|
||||
expect(res.status).toBe(400);
|
||||
expect(res.body?.code).toBe("SCHEMA_ERROR");
|
||||
expect(res.body?.error).toContain("Custom event data must be a JSON object");
|
||||
});
|
||||
|
||||
it("rejects $-prefixed span types", async ({ expect }) => {
|
||||
await setupAnalyticsProject();
|
||||
await Auth.Otp.signIn();
|
||||
|
||||
const res = await uploadTelemetryBatch({
|
||||
session_replay_segment_id: randomUUID(),
|
||||
spans: [makeCustomSpan({ span_type: "$reserved" })],
|
||||
});
|
||||
|
||||
expect(res.status).toBe(400);
|
||||
expect(res.body?.code).toBe("SCHEMA_ERROR");
|
||||
expect(res.body?.error).toContain("Invalid span_type");
|
||||
});
|
||||
|
||||
it("rejects spans that end before they start", async ({ expect }) => {
|
||||
await setupAnalyticsProject();
|
||||
await Auth.Otp.signIn();
|
||||
|
||||
const now = Date.now();
|
||||
const res = await uploadTelemetryBatch({
|
||||
session_replay_segment_id: randomUUID(),
|
||||
spans: [makeCustomSpan({ started_at_ms: now, ended_at_ms: now - 1000 })],
|
||||
});
|
||||
|
||||
expect(res.status).toBe(400);
|
||||
expect(res.body?.code).toBe("SCHEMA_ERROR");
|
||||
expect(res.body?.error).toContain("ended_at_ms must be greater than or equal to started_at_ms");
|
||||
});
|
||||
|
||||
it("rejects a batch with neither events nor spans", async ({ expect }) => {
|
||||
await setupAnalyticsProject();
|
||||
await Auth.Otp.signIn();
|
||||
|
||||
const res = await uploadTelemetryBatch({
|
||||
session_replay_segment_id: randomUUID(),
|
||||
});
|
||||
|
||||
expect(res.status).toBe(400);
|
||||
expect(res.body?.code).toBe("SCHEMA_ERROR");
|
||||
expect(res.body?.error).toContain("A batch must contain at least one event or span");
|
||||
});
|
||||
|
||||
it("rejects user_id with client auth", async ({ expect }) => {
|
||||
await setupAnalyticsProject();
|
||||
const { userId } = await Auth.Otp.signIn();
|
||||
|
||||
const res = await uploadTelemetryBatch({
|
||||
session_replay_segment_id: randomUUID(),
|
||||
user_id: userId,
|
||||
events: [{ event_type: "checkout_completed", event_at_ms: Date.now(), data: {} }],
|
||||
});
|
||||
|
||||
expect(res.status).toBe(400);
|
||||
expect(res.body).toBe("user_id must not be set with client auth; it is derived from the session");
|
||||
});
|
||||
|
||||
it("rejects client-auth batches without session_replay_segment_id", async ({ expect }) => {
|
||||
await setupAnalyticsProject();
|
||||
await Auth.Otp.signIn();
|
||||
|
||||
const res = await uploadTelemetryBatch({
|
||||
events: [{ event_type: "checkout_completed", event_at_ms: Date.now(), data: {} }],
|
||||
});
|
||||
|
||||
expect(res.status).toBe(400);
|
||||
expect(res.body).toBe("session_replay_segment_id is required for analytics batches with client auth");
|
||||
});
|
||||
|
||||
it("accepts a spans-only batch and lands it on the spans surface", async ({ expect }) => {
|
||||
await setupAnalyticsProject();
|
||||
const { userId } = await Auth.Otp.signIn();
|
||||
|
||||
const sessionReplaySegmentId = randomUUID();
|
||||
const spanId = randomUUID();
|
||||
const now = Date.now();
|
||||
const res = await uploadTelemetryBatch({
|
||||
session_replay_segment_id: sessionReplaySegmentId,
|
||||
spans: [{
|
||||
span_id: spanId,
|
||||
span_type: "checkout-flow",
|
||||
started_at_ms: now - 1000,
|
||||
ended_at_ms: null,
|
||||
parent_span_ids: [],
|
||||
data: {},
|
||||
updated_at_ms: now,
|
||||
}],
|
||||
});
|
||||
expect(res).toMatchInlineSnapshot(`
|
||||
NiceResponse {
|
||||
"status": 200,
|
||||
"body": { "inserted": 1 },
|
||||
"headers": Headers { <some fields may have been hidden> },
|
||||
}
|
||||
`);
|
||||
|
||||
const queryRes = await queryAnalyticsUntil({
|
||||
query: "SELECT id, span_type, span_ended_at, parent_span_ids, user_id, session_replay_segment_id FROM spans WHERE id = {id:String}",
|
||||
params: { id: `cs-${spanId}` },
|
||||
}, (r) => r.body?.result?.length === 1);
|
||||
|
||||
expect(queryRes?.status).toBe(200);
|
||||
const rows = (queryRes?.body as any).result;
|
||||
expect(rows).toHaveLength(1);
|
||||
expect(rows[0].id).toBe(`cs-${spanId}`);
|
||||
expect(rows[0].span_type).toBe("checkout-flow");
|
||||
// The span is still open.
|
||||
expect(rows[0].span_ended_at).toBeNull();
|
||||
// No active session replay, so the only system ancestor is the refresh-token span.
|
||||
expect(rows[0].parent_span_ids).toHaveLength(1);
|
||||
expect(rows[0].parent_span_ids[0]).toMatch(/^rti-/);
|
||||
expect(rows[0].user_id).toBe(userId);
|
||||
expect(rows[0].session_replay_segment_id).toBe(sessionReplaySegmentId);
|
||||
});
|
||||
|
||||
it("collapses open→closed span re-writes to the ended row", async ({ expect }) => {
|
||||
await setupAnalyticsProject();
|
||||
await Auth.Otp.signIn();
|
||||
|
||||
const sessionReplaySegmentId = randomUUID();
|
||||
const spanId = randomUUID();
|
||||
const now = Date.now();
|
||||
const startedAtMs = now - 10_000;
|
||||
const endedAtMs = now - 2000;
|
||||
const openUpdatedAtMs = now - 9000;
|
||||
const closedUpdatedAtMs = now - 1000;
|
||||
|
||||
const openRes = await uploadTelemetryBatch({
|
||||
session_replay_segment_id: sessionReplaySegmentId,
|
||||
spans: [makeCustomSpan({ span_id: spanId, started_at_ms: startedAtMs, ended_at_ms: null, updated_at_ms: openUpdatedAtMs })],
|
||||
});
|
||||
expect(openRes.status).toBe(200);
|
||||
|
||||
const openQueryRes = await queryAnalyticsUntil({
|
||||
query: "SELECT id, span_ended_at FROM spans WHERE id = {id:String}",
|
||||
params: { id: `cs-${spanId}` },
|
||||
}, (r) => r.body?.result?.length === 1);
|
||||
expect(openQueryRes?.status).toBe(200);
|
||||
expect((openQueryRes?.body as any).result[0].span_ended_at).toBeNull();
|
||||
|
||||
const closedRes = await uploadTelemetryBatch({
|
||||
session_replay_segment_id: sessionReplaySegmentId,
|
||||
spans: [makeCustomSpan({ span_id: spanId, started_at_ms: startedAtMs, ended_at_ms: endedAtMs, updated_at_ms: closedUpdatedAtMs })],
|
||||
});
|
||||
expect(closedRes.status).toBe(200);
|
||||
|
||||
// The view reads FINAL, so the two versions collapse to the ended row.
|
||||
const closedQueryRes = await queryAnalyticsUntil({
|
||||
query: "SELECT id, span_ended_at FROM spans WHERE id = {id:String}",
|
||||
params: { id: `cs-${spanId}` },
|
||||
}, (r) => r.body?.result?.length === 1 && r.body.result[0].span_ended_at != null);
|
||||
expect(closedQueryRes?.status).toBe(200);
|
||||
const rows = (closedQueryRes?.body as any).result;
|
||||
expect(rows).toHaveLength(1);
|
||||
expect(rows[0].span_ended_at).not.toBeNull();
|
||||
});
|
||||
|
||||
it("keeps the ended row when an open re-write arrives with an older version (out-of-order)", async ({ expect }) => {
|
||||
await setupAnalyticsProject();
|
||||
await Auth.Otp.signIn();
|
||||
|
||||
const sessionReplaySegmentId = randomUUID();
|
||||
const spanId = randomUUID();
|
||||
const sentinelSpanId = randomUUID();
|
||||
const now = Date.now();
|
||||
const startedAtMs = now - 10_000;
|
||||
const endedAtMs = now - 2000;
|
||||
const openUpdatedAtMs = now - 9000;
|
||||
const closedUpdatedAtMs = now - 1000;
|
||||
|
||||
// The END row arrives first, carrying the LATER version…
|
||||
const closedRes = await uploadTelemetryBatch({
|
||||
session_replay_segment_id: sessionReplaySegmentId,
|
||||
spans: [makeCustomSpan({ span_id: spanId, started_at_ms: startedAtMs, ended_at_ms: endedAtMs, updated_at_ms: closedUpdatedAtMs })],
|
||||
});
|
||||
expect(closedRes.status).toBe(200);
|
||||
|
||||
const closedQueryRes = await queryAnalyticsUntil({
|
||||
query: "SELECT id, span_ended_at FROM spans WHERE id = {id:String}",
|
||||
params: { id: `cs-${spanId}` },
|
||||
}, (r) => r.body?.result?.length === 1);
|
||||
expect(closedQueryRes?.status).toBe(200);
|
||||
expect((closedQueryRes?.body as any).result[0].span_ended_at).not.toBeNull();
|
||||
|
||||
// …then a stale OPEN row with the EARLIER version arrives. The sentinel span
|
||||
// rides in the same batch (same ClickHouse insert), so once it is visible the
|
||||
// stale open row has landed too.
|
||||
const staleOpenRes = await uploadTelemetryBatch({
|
||||
session_replay_segment_id: sessionReplaySegmentId,
|
||||
spans: [
|
||||
makeCustomSpan({ span_id: spanId, started_at_ms: startedAtMs, ended_at_ms: null, updated_at_ms: openUpdatedAtMs }),
|
||||
makeCustomSpan({ span_id: sentinelSpanId }),
|
||||
],
|
||||
});
|
||||
expect(staleOpenRes.status).toBe(200);
|
||||
|
||||
const sentinelQueryRes = await queryAnalyticsUntil({
|
||||
query: "SELECT id FROM spans WHERE id = {id:String}",
|
||||
params: { id: `cs-${sentinelSpanId}` },
|
||||
}, (r) => r.body?.result?.length === 1);
|
||||
expect(sentinelQueryRes?.status).toBe(200);
|
||||
|
||||
// The ended row still wins: version (updated_at_ms) decides, not insert order.
|
||||
const finalQueryRes = await queryAnalyticsUntil({
|
||||
query: "SELECT id, span_ended_at FROM spans WHERE id = {id:String}",
|
||||
params: { id: `cs-${spanId}` },
|
||||
}, (r) => r.body?.result?.length === 1);
|
||||
expect(finalQueryRes?.status).toBe(200);
|
||||
const rows = (finalQueryRes?.body as any).result;
|
||||
expect(rows).toHaveLength(1);
|
||||
expect(rows[0].span_ended_at).not.toBeNull();
|
||||
});
|
||||
|
||||
it("accepts server-key batches with explicit user_id and no system ancestry", async ({ expect }) => {
|
||||
await setupAnalyticsProject();
|
||||
const { userId } = await Auth.Otp.signIn();
|
||||
// Server-key request: no user access token, no refresh token.
|
||||
backendContext.set({ userAuth: null });
|
||||
|
||||
const eventType = `sk-${randomUUID()}`;
|
||||
const res = await uploadTelemetryBatch({
|
||||
user_id: userId,
|
||||
events: [{ event_type: eventType, event_at_ms: Date.now(), data: { source: "server" } }],
|
||||
}, { accessType: "server" });
|
||||
expect(res).toMatchInlineSnapshot(`
|
||||
NiceResponse {
|
||||
"status": 200,
|
||||
"body": { "inserted": 1 },
|
||||
"headers": Headers { <some fields may have been hidden> },
|
||||
}
|
||||
`);
|
||||
|
||||
const queryRes = await queryAnalyticsUntil({
|
||||
query: "SELECT user_id, parent_span_ids FROM events WHERE event_type = {eventType:String}",
|
||||
params: { eventType },
|
||||
}, (r) => r.body?.result?.length === 1);
|
||||
|
||||
expect(queryRes?.status).toBe(200);
|
||||
const row = (queryRes?.body as any).result[0];
|
||||
expect(row.user_id).toBe(userId);
|
||||
// No session on server auth, so there is no system ancestry at all.
|
||||
expect(row.parent_span_ids).toEqual([]);
|
||||
});
|
||||
|
||||
it("rejects server-key batches with an unknown user_id", async ({ expect }) => {
|
||||
await setupAnalyticsProject();
|
||||
backendContext.set({ userAuth: null });
|
||||
|
||||
const res = await uploadTelemetryBatch({
|
||||
user_id: randomUUID(),
|
||||
events: [{ event_type: "checkout_completed", event_at_ms: Date.now(), data: {} }],
|
||||
}, { accessType: "server" });
|
||||
|
||||
expect(res.status).toBe(400);
|
||||
expect(res.body).toBe("user_id does not correspond to a user on this project/branch");
|
||||
});
|
||||
|
||||
it("accepts server-key spans without refresh-token ancestry", async ({ expect }) => {
|
||||
await setupAnalyticsProject();
|
||||
const { userId } = await Auth.Otp.signIn();
|
||||
backendContext.set({ userAuth: null });
|
||||
|
||||
const spanId = randomUUID();
|
||||
const res = await uploadTelemetryBatch({
|
||||
user_id: userId,
|
||||
spans: [makeCustomSpan({ span_id: spanId })],
|
||||
}, { accessType: "server" });
|
||||
expect(res.status).toBe(200);
|
||||
expect(res.body.inserted).toBe(1);
|
||||
|
||||
const queryRes = await queryAnalyticsUntil({
|
||||
query: "SELECT id, span_type, parent_span_ids, user_id FROM spans WHERE id = {id:String}",
|
||||
params: { id: `cs-${spanId}` },
|
||||
}, (r) => r.body?.result?.length === 1);
|
||||
|
||||
expect(queryRes?.status).toBe(200);
|
||||
const rows = (queryRes?.body as any).result;
|
||||
expect(rows).toHaveLength(1);
|
||||
expect(rows[0].span_type).toBe("checkout-flow");
|
||||
expect(rows[0].user_id).toBe(userId);
|
||||
// No refresh token on server auth → no rti- (or any system) ancestor.
|
||||
expect(rows[0].parent_span_ids).toEqual([]);
|
||||
});
|
||||
|
||||
it("accepts a gzipped binary body containing custom events and spans", async ({ expect }) => {
|
||||
await setupAnalyticsProject();
|
||||
await Auth.Otp.signIn();
|
||||
|
||||
const now = Date.now();
|
||||
const payload = {
|
||||
session_replay_segment_id: randomUUID(),
|
||||
batch_id: randomUUID(),
|
||||
sent_at_ms: now,
|
||||
events: [{ event_type: "checkout_completed", event_at_ms: now - 100, data: { cart_size: 3 } }],
|
||||
spans: [makeCustomSpan()],
|
||||
};
|
||||
const compressed = gzipSync(Buffer.from(JSON.stringify(payload), "utf-8"));
|
||||
|
||||
const res = await niceBackendFetch("/api/v1/analytics/events/batch", {
|
||||
method: "POST",
|
||||
accessType: "client",
|
||||
rawBody: compressed,
|
||||
});
|
||||
|
||||
expect(res).toMatchInlineSnapshot(`
|
||||
NiceResponse {
|
||||
"status": 200,
|
||||
"body": { "inserted": 2 },
|
||||
"headers": Headers { <some fields may have been hidden> },
|
||||
}
|
||||
`);
|
||||
});
|
||||
|
||||
@ -130,7 +130,7 @@ function getBotChallengeRequestFields(botChallenge: BotChallengeInput | undefine
|
||||
};
|
||||
}
|
||||
|
||||
async function encodeGzipJsonBody(
|
||||
export async function encodeGzipJsonBody(
|
||||
jsonBody: string,
|
||||
options: { keepalive: boolean },
|
||||
): Promise<{ body: BodyInit, contentType: string }> {
|
||||
|
||||
@ -9,6 +9,7 @@ import { Result } from "../utils/results";
|
||||
import { urlString } from "../utils/urls";
|
||||
import {
|
||||
ClientInterfaceOptions,
|
||||
encodeGzipJsonBody,
|
||||
HexclaveClientInterface
|
||||
} from "./client-interface";
|
||||
import { ConnectedAccountAccessTokenCrud, ConnectedAccountCrud } from "./crud/connected-accounts";
|
||||
@ -59,6 +60,38 @@ export class HexclaveServerInterface extends HexclaveClientInterface {
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Server-auth variant of the analytics batch upload (see
|
||||
* `sendAnalyticsEventBatch` on the client interface): authenticates with the
|
||||
* secret server key instead of a user session, against the analytics base URL.
|
||||
* Single attempt, like the client path — analytics is fire-and-forget and the
|
||||
* caller's returned promise carries the outcome.
|
||||
*/
|
||||
async sendAnalyticsEventBatchAsServer(body: string): Promise<Result<Response, Error>> {
|
||||
try {
|
||||
const encoded = await encodeGzipJsonBody(body, { keepalive: false });
|
||||
const response = await this.sendClientRequest(
|
||||
"/analytics/events/batch",
|
||||
{
|
||||
method: "POST",
|
||||
headers: {
|
||||
"Content-Type": encoded.contentType,
|
||||
// Hexclave rebrand: emit x-hexclave-* request header; the backend proxy dual-accepts both names.
|
||||
"x-hexclave-secret-server-key": "secretServerKey" in this.options ? this.options.secretServerKey : "",
|
||||
},
|
||||
body: encoded.body,
|
||||
},
|
||||
null,
|
||||
"server",
|
||||
this.getAnalyticsApiUrl(),
|
||||
{ maxAttempts: 1, skipDiagnostics: true },
|
||||
);
|
||||
return Result.ok(response);
|
||||
} catch (e) {
|
||||
return Result.error(e instanceof Error ? e : new Error(String(e)));
|
||||
}
|
||||
}
|
||||
|
||||
override async getCustomerBilling(
|
||||
customerType: "user" | "team",
|
||||
customerId: string,
|
||||
|
||||
@ -6,6 +6,10 @@ export { getConvexProvidersConfig } from "./integrations/convex";
|
||||
export type { HexclaveConfig, StackConfig } from "@hexclave/shared/config";
|
||||
export { defineHexclaveConfig, defineStackConfig } from "@hexclave/shared/config";
|
||||
|
||||
// Custom telemetry (trackEvent/startSpan) — platform-neutral: the methods exist
|
||||
// on every SDK surface (non-browser environments no-op with inert spans).
|
||||
export type { ParentRef, Span, SpanRef, StartSpanOptions, TrackOptions } from "./lib/hexclave-app/apps/implementations/event-tracker";
|
||||
|
||||
// IF_PLATFORM react-like
|
||||
export type { AnalyticsOptions, AnalyticsReplayOptions } from "./lib/hexclave-app/apps/implementations/session-replay";
|
||||
// Hexclave aliases and legacy Stack* names — @deprecated JSDoc lives on the original
|
||||
|
||||
@ -61,7 +61,7 @@ import { ActiveSession, Auth, BaseUser, CurrentUser, InternalUserExtra, OAuthPro
|
||||
import { StackClientApp, StackClientAppConstructorOptions, StackClientAppJson } from "../interfaces/client-app";
|
||||
import { _HexclaveAdminAppImplIncomplete } from "./admin-app-impl";
|
||||
import { TokenObject, clientVersion, createCache, createCacheBySession, createEmptyTokenStore, getAnalyticsBaseUrl, getDefaultExtraRequestHeaders, getDefaultProjectId, getDefaultPublishableClientKey, getUrls, resolveApiUrls, resolveConstructorOptions } from "./common";
|
||||
import { EventTracker } from "./event-tracker";
|
||||
import { createInertSpan, EventTracker, getCustomTelemetryDataError, getCustomTelemetryNameError, rejectedPreCaught, warnTelemetryUnavailableOnce, type Span, type StartSpanOptions, type TrackOptions } from "./event-tracker";
|
||||
import type { CrossDomainHandoffParams } from "./redirect-page-urls";
|
||||
import { crossDomainAuthQueryParams, getCrossDomainHandoffParamsFromCurrentUrl, planRedirectToHandler } from "./redirect-page-urls";
|
||||
import { subscribeSessionRefresh } from "./session-refresh-subscription";
|
||||
@ -258,7 +258,7 @@ export class _HexclaveClientAppImplIncomplete<HasTokenStore extends boolean, Pro
|
||||
|
||||
private readonly _analyticsOptions: AnalyticsOptions | undefined;
|
||||
private _sessionRecorder: SessionRecorder | null = null;
|
||||
private _eventTracker: EventTracker | null = null;
|
||||
protected _eventTracker: EventTracker | null = null;
|
||||
|
||||
private __DEMO_ENABLE_SLIGHT_FETCH_DELAY = false;
|
||||
private readonly _ownedAdminApps = new DependenciesMap<[InternalSession, string], _HexclaveAdminAppImplIncomplete<false, string>>();
|
||||
@ -3995,6 +3995,48 @@ export class _HexclaveClientAppImplIncomplete<HasTokenStore extends boolean, Pro
|
||||
}
|
||||
}
|
||||
|
||||
// Custom telemetry (see the StackClientApp interface for user-facing docs).
|
||||
// The tracker only exists in browser-like environments with analytics enabled
|
||||
// and a persistent token store; everywhere else these are validated no-ops so
|
||||
// isomorphic code never needs to branch (and misuse still surfaces loudly).
|
||||
|
||||
trackEvent(eventType: string, data?: Record<string, unknown>, options?: TrackOptions): Promise<void> {
|
||||
if (this._eventTracker) {
|
||||
return this._eventTracker.trackCustomEvent(eventType, data, options);
|
||||
}
|
||||
const nameError = getCustomTelemetryNameError("event", eventType);
|
||||
if (nameError) return rejectedPreCaught(nameError);
|
||||
const dataError = getCustomTelemetryDataError(data);
|
||||
if (dataError) return rejectedPreCaught(dataError);
|
||||
warnTelemetryUnavailableOnce();
|
||||
return Promise.resolve();
|
||||
}
|
||||
|
||||
startSpan(spanType: string, options?: StartSpanOptions): Span {
|
||||
if (this._eventTracker) {
|
||||
return this._eventTracker.startSpan(spanType, options);
|
||||
}
|
||||
const nameError = getCustomTelemetryNameError("span", spanType);
|
||||
if (nameError) {
|
||||
console.error(`Hexclave analytics: ${nameError}`);
|
||||
} else {
|
||||
warnTelemetryUnavailableOnce();
|
||||
}
|
||||
return createInertSpan(spanType);
|
||||
}
|
||||
|
||||
setGlobalSpan(span: Span): void {
|
||||
this._eventTracker?.setGlobalSpan(span);
|
||||
}
|
||||
|
||||
unsetGlobalSpan(span: Span): void {
|
||||
this._eventTracker?.unsetGlobalSpan(span);
|
||||
}
|
||||
|
||||
async flush(): Promise<void> {
|
||||
await this._eventTracker?.flush();
|
||||
}
|
||||
|
||||
async getAccessToken(options?: { tokenStore?: TokenStoreInit }): Promise<string | null> {
|
||||
const user = await this.getUser({ tokenStore: options?.tokenStore ?? undefined as any });
|
||||
if (user) {
|
||||
|
||||
@ -420,6 +420,320 @@ describe("EventTracker", () => {
|
||||
}
|
||||
});
|
||||
|
||||
it("buffers custom events alongside system events and resolves their promises on ack", async () => {
|
||||
vi.useFakeTimers();
|
||||
document.body.innerHTML = "";
|
||||
|
||||
const sentBodies: string[] = [];
|
||||
const tracker = new EventTracker({
|
||||
projectId: "internal",
|
||||
sendBatch: async (body) => {
|
||||
sentBodies.push(body);
|
||||
return Result.ok(new Response());
|
||||
},
|
||||
});
|
||||
|
||||
try {
|
||||
tracker.start();
|
||||
const promise = tracker.trackCustomEvent("checkout_completed", { cart_size: 3 });
|
||||
|
||||
await advancePastFlush();
|
||||
await expect(promise).resolves.toBeUndefined();
|
||||
|
||||
const payload = JSON.parse(sentBodies[0] ?? "{}") as { events: { event_type: string, data: Record<string, unknown>, parent_span_ids?: string[] }[] };
|
||||
expect(payload.events.map((event) => event.event_type)).toEqual(["$page-view", "checkout_completed"]);
|
||||
const custom = payload.events[1];
|
||||
expect(custom.data).toEqual({ cart_size: 3 });
|
||||
// No parents were given and no globals are set — the key is omitted
|
||||
// entirely (the server stamps system ancestry on every event anyway).
|
||||
expect(custom.parent_span_ids).toBeUndefined();
|
||||
} finally {
|
||||
tracker.stop();
|
||||
}
|
||||
});
|
||||
|
||||
it("rejects (pre-caught, never throws) on invalid names, data, and parent ids", async () => {
|
||||
vi.useFakeTimers();
|
||||
const errorSpy = vi.spyOn(console, "error").mockImplementation(() => {});
|
||||
const tracker = new EventTracker({
|
||||
projectId: "internal",
|
||||
sendBatch: async () => Result.ok(new Response()),
|
||||
});
|
||||
|
||||
try {
|
||||
tracker.start();
|
||||
// $-prefixed names are reserved for system telemetry.
|
||||
await expect(tracker.trackCustomEvent("$page-view")).rejects.toThrow(/reserved for system telemetry/);
|
||||
await expect(tracker.trackCustomEvent("1-starts-with-digit")).rejects.toThrow(/must start with a letter/);
|
||||
await expect(tracker.trackCustomEvent("x".repeat(65))).rejects.toThrow(/at most 64 characters/);
|
||||
await expect(tracker.trackCustomEvent("ok", [1, 2] as any)).rejects.toThrow(/plain JSON-serializable object/);
|
||||
await expect(tracker.trackCustomEvent("ok", { big: "x".repeat(17_000) })).rejects.toThrow(/at most 16000 bytes/);
|
||||
await expect(tracker.trackCustomEvent("ok", {}, { parentIds: ["not-a-uuid"] })).rejects.toThrow(/parent ids must be span uuids/);
|
||||
// Ignoring the rejected promise entirely must not blow up the test run
|
||||
// (the internal catch keeps it from ever being an unhandled rejection).
|
||||
tracker.trackCustomEvent("$ignored").catch(() => {});
|
||||
expect(errorSpy).toHaveBeenCalled();
|
||||
|
||||
// Invalid startSpan input yields an inert span rather than a throw.
|
||||
const inert = tracker.startSpan("$reserved");
|
||||
await expect(inert.end()).resolves.toBeUndefined();
|
||||
|
||||
// Nothing invalid was buffered.
|
||||
await advancePastFlush();
|
||||
} finally {
|
||||
tracker.stop();
|
||||
errorSpy.mockRestore();
|
||||
}
|
||||
});
|
||||
|
||||
it("writes a span open on start, dedupes in-batch re-writes to the latest row, and settles every promise", async () => {
|
||||
vi.useFakeTimers();
|
||||
const sentBodies: string[] = [];
|
||||
const tracker = new EventTracker({
|
||||
projectId: "internal",
|
||||
sendBatch: async (body) => {
|
||||
sentBodies.push(body);
|
||||
return Result.ok(new Response());
|
||||
},
|
||||
});
|
||||
|
||||
try {
|
||||
tracker.start();
|
||||
const span = tracker.startSpan("checkout-flow", { data: { cart_size: 3 } });
|
||||
const setDataPromise = span.setData({ coupon: "SAVE10" });
|
||||
const endPromise = span.end();
|
||||
expect(span.isEnded).toBe(true);
|
||||
// end() is idempotent: repeat calls return the first call's promise.
|
||||
expect(span.end()).toBe(endPromise);
|
||||
|
||||
await advancePastFlush();
|
||||
await expect(setDataPromise).resolves.toBeUndefined();
|
||||
await expect(endPromise).resolves.toBeUndefined();
|
||||
|
||||
// Start + setData + end within one flush window: exactly ONE wire row,
|
||||
// carrying the latest state (ended, merged data).
|
||||
const payload = JSON.parse(sentBodies[0] ?? "{}") as { spans?: { span_id: string, span_type: string, ended_at_ms: number | null, data: Record<string, unknown>, parent_span_ids: string[] }[] };
|
||||
expect(payload.spans).toHaveLength(1);
|
||||
const row = payload.spans![0];
|
||||
expect(row.span_id).toBe(span.spanId);
|
||||
expect(row.span_type).toBe("checkout-flow");
|
||||
expect(row.ended_at_ms).not.toBeNull();
|
||||
expect(row.data).toEqual({ cart_size: 3, coupon: "SAVE10" });
|
||||
expect(row.parent_span_ids).toEqual([]);
|
||||
} finally {
|
||||
tracker.stop();
|
||||
}
|
||||
});
|
||||
|
||||
it("re-writes a span across flushes with a strictly increasing version", async () => {
|
||||
vi.useFakeTimers();
|
||||
const sentBodies: string[] = [];
|
||||
const tracker = new EventTracker({
|
||||
projectId: "internal",
|
||||
sendBatch: async (body) => {
|
||||
sentBodies.push(body);
|
||||
return Result.ok(new Response());
|
||||
},
|
||||
});
|
||||
|
||||
try {
|
||||
tracker.start();
|
||||
const span = tracker.startSpan("upload");
|
||||
await advancePastFlush();
|
||||
|
||||
span.end().catch(() => {});
|
||||
await advancePastFlush();
|
||||
|
||||
const first = JSON.parse(sentBodies[0] ?? "{}") as { spans?: { ended_at_ms: number | null, updated_at_ms: number }[] };
|
||||
const second = JSON.parse(sentBodies[1] ?? "{}") as { spans?: { ended_at_ms: number | null, updated_at_ms: number }[] };
|
||||
expect(first.spans![0].ended_at_ms).toBeNull();
|
||||
expect(second.spans![0].ended_at_ms).not.toBeNull();
|
||||
// The end row must always beat the open row in the ReplacingMergeTree,
|
||||
// even if the batches arrive out of order server-side.
|
||||
expect(second.spans![0].updated_at_ms).toBeGreaterThan(first.spans![0].updated_at_ms);
|
||||
} finally {
|
||||
tracker.stop();
|
||||
}
|
||||
});
|
||||
|
||||
it("parents children and events under global spans and handle chains (span2.trackEvent inherits everything)", async () => {
|
||||
vi.useFakeTimers();
|
||||
const sentBodies: string[] = [];
|
||||
const tracker = new EventTracker({
|
||||
projectId: "internal",
|
||||
sendBatch: async (body) => {
|
||||
sentBodies.push(body);
|
||||
return Result.ok(new Response());
|
||||
},
|
||||
});
|
||||
|
||||
try {
|
||||
tracker.start();
|
||||
const checkout = tracker.startSpan("checkout-flow");
|
||||
tracker.setGlobalSpan(checkout);
|
||||
|
||||
// Child created via the handle: chain = [checkout].
|
||||
const payment = checkout.startSpan("payment-step");
|
||||
// Event tracked via the child handle: parents = ambient (checkout) +
|
||||
// payment's chain (checkout, deduped) + payment itself.
|
||||
const eventPromise = payment.trackEvent("card_declined", { code: "51" });
|
||||
|
||||
// A raw uuid parent contributes only itself, additively.
|
||||
const rawParent = "0f000000-0000-4000-8000-00000000cccc";
|
||||
tracker.trackCustomEvent("hint_shown", {}, { parentIds: [rawParent] }).catch(() => {});
|
||||
|
||||
await advancePastFlush();
|
||||
await expect(eventPromise).resolves.toBeUndefined();
|
||||
|
||||
const payload = JSON.parse(sentBodies[0] ?? "{}") as {
|
||||
events: { event_type: string, parent_span_ids?: string[] }[],
|
||||
spans?: { span_id: string, parent_span_ids: string[] }[],
|
||||
};
|
||||
const paymentRow = payload.spans!.find((row) => row.span_id === payment.spanId);
|
||||
expect(paymentRow!.parent_span_ids).toEqual([checkout.spanId]);
|
||||
|
||||
const declined = payload.events.find((event) => event.event_type === "card_declined");
|
||||
expect(declined!.parent_span_ids).toEqual([checkout.spanId, payment.spanId]);
|
||||
|
||||
const hint = payload.events.find((event) => event.event_type === "hint_shown");
|
||||
expect(hint!.parent_span_ids).toEqual([checkout.spanId, rawParent]);
|
||||
|
||||
// Ending a global span auto-unsets it: subsequent events have no parents.
|
||||
checkout.end().catch(() => {});
|
||||
tracker.trackCustomEvent("after_end").catch(() => {});
|
||||
await advancePastFlush();
|
||||
const second = JSON.parse(sentBodies[1] ?? "{}") as { events: { event_type: string, parent_span_ids?: string[] }[] };
|
||||
const after = second.events.find((event) => event.event_type === "after_end");
|
||||
expect(after!.parent_span_ids).toBeUndefined();
|
||||
} finally {
|
||||
tracker.stop();
|
||||
}
|
||||
});
|
||||
|
||||
it("continues a span tree from a serialized SpanRef with full ancestry", async () => {
|
||||
vi.useFakeTimers();
|
||||
const sentBodies: string[] = [];
|
||||
const tracker = new EventTracker({
|
||||
projectId: "internal",
|
||||
sendBatch: async (body) => {
|
||||
sentBodies.push(body);
|
||||
return Result.ok(new Response());
|
||||
},
|
||||
});
|
||||
|
||||
try {
|
||||
tracker.start();
|
||||
// Simulates a span minted on another tier (e.g. the server) and passed
|
||||
// through JSON: the ref carries its own ancestor chain.
|
||||
const serverRef = {
|
||||
spanId: "0f000000-0000-4000-8000-00000000dddd",
|
||||
parentSpanIds: ["0f000000-0000-4000-8000-00000000eeee"],
|
||||
};
|
||||
const child = tracker.startSpan("client-continuation", { parentIds: [serverRef] });
|
||||
await advancePastFlush();
|
||||
|
||||
const payload = JSON.parse(sentBodies[0] ?? "{}") as { spans?: { span_id: string, parent_span_ids: string[] }[] };
|
||||
expect(payload.spans![0].span_id).toBe(child.spanId);
|
||||
expect(payload.spans![0].parent_span_ids).toEqual([serverRef.parentSpanIds[0], serverRef.spanId]);
|
||||
expect(child.ref()).toEqual({ spanId: child.spanId, parentSpanIds: [serverRef.parentSpanIds[0], serverRef.spanId] });
|
||||
} finally {
|
||||
tracker.stop();
|
||||
}
|
||||
});
|
||||
|
||||
it("rejects pending promises on failed sends without unhandled rejections, and flush() sends immediately", async () => {
|
||||
vi.useFakeTimers();
|
||||
const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {});
|
||||
let failNext = true;
|
||||
const sentBodies: string[] = [];
|
||||
const tracker = new EventTracker({
|
||||
projectId: "internal",
|
||||
sendBatch: async (body) => {
|
||||
sentBodies.push(body);
|
||||
return failNext ? Result.error(new TypeError("Failed to fetch")) : Result.ok(new Response());
|
||||
},
|
||||
});
|
||||
|
||||
try {
|
||||
tracker.start();
|
||||
const failed = tracker.trackCustomEvent("will_fail");
|
||||
// flush() is the "send now" escape hatch — no timer advance needed.
|
||||
await tracker.flush();
|
||||
expect(sentBodies).toHaveLength(1);
|
||||
await expect(failed).rejects.toThrow("Failed to fetch");
|
||||
|
||||
failNext = false;
|
||||
const succeeded = tracker.trackCustomEvent("will_succeed");
|
||||
await tracker.flush();
|
||||
await expect(succeeded).resolves.toBeUndefined();
|
||||
} finally {
|
||||
tracker.stop();
|
||||
warnSpy.mockRestore();
|
||||
}
|
||||
});
|
||||
|
||||
it("clearBuffer drops pending telemetry and inert-ifies live spans (sign-out privacy)", async () => {
|
||||
vi.useFakeTimers();
|
||||
const sentBodies: string[] = [];
|
||||
const tracker = new EventTracker({
|
||||
projectId: "internal",
|
||||
sendBatch: async (body) => {
|
||||
sentBodies.push(body);
|
||||
return Result.ok(new Response());
|
||||
},
|
||||
});
|
||||
|
||||
try {
|
||||
tracker.start();
|
||||
const span = tracker.startSpan("cross-user-flow");
|
||||
const pendingEvent = tracker.trackCustomEvent("pre_signout");
|
||||
|
||||
tracker.clearBuffer();
|
||||
await expect(pendingEvent).rejects.toThrow(/buffer cleared/);
|
||||
|
||||
// A post-clear end() must not write anything: the handle is inert, so a
|
||||
// span started by user A can never be re-written under user B.
|
||||
await expect(span.end()).resolves.toBeUndefined();
|
||||
await advancePastFlush();
|
||||
|
||||
for (const body of sentBodies) {
|
||||
const payload = JSON.parse(body) as { events: { event_type: string }[], spans?: unknown[] };
|
||||
expect(payload.spans).toBeUndefined();
|
||||
expect(payload.events.every((event) => event.event_type === "$page-view")).toBe(true);
|
||||
}
|
||||
} finally {
|
||||
tracker.stop();
|
||||
}
|
||||
});
|
||||
|
||||
it("ships buffered spans on the keepalive pagehide flush (open spans stay open)", async () => {
|
||||
vi.useFakeTimers();
|
||||
const sentBodies: string[] = [];
|
||||
const tracker = new EventTracker({
|
||||
projectId: "internal",
|
||||
sendBatch: async (body) => {
|
||||
sentBodies.push(body);
|
||||
return Result.ok(new Response());
|
||||
},
|
||||
});
|
||||
|
||||
try {
|
||||
tracker.start();
|
||||
tracker.startSpan("abandoned-flow");
|
||||
window.dispatchEvent(new Event("pagehide"));
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
|
||||
const payload = JSON.parse(sentBodies[0] ?? "{}") as { spans?: { span_type: string, ended_at_ms: number | null }[] };
|
||||
expect(payload.spans).toHaveLength(1);
|
||||
expect(payload.spans![0].span_type).toBe("abandoned-flow");
|
||||
// No auto-end on unload: the span survives as an open interval by design.
|
||||
expect(payload.spans![0].ended_at_ms).toBeNull();
|
||||
} finally {
|
||||
tracker.stop();
|
||||
}
|
||||
});
|
||||
|
||||
it("silently disables when client interface returns ANALYTICS_NOT_ENABLED as an error", async () => {
|
||||
vi.useFakeTimers();
|
||||
document.body.innerHTML = "<button>Click me</button>";
|
||||
|
||||
@ -10,6 +10,206 @@ const FLUSH_INTERVAL_MS = 10_000;
|
||||
const MAX_EVENTS_PER_BATCH = 50;
|
||||
const MAX_APPROX_BYTES_PER_BATCH = 64_000;
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Custom telemetry (public trackEvent/startSpan API)
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
// 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 server-side validation in apps/backend's analytics events batch route.
|
||||
const CUSTOM_TELEMETRY_NAME_RE = /^[a-zA-Z][a-zA-Z0-9_.:-]{0,63}$/;
|
||||
// Serialized size cap per event/span data payload; mirrors the server cap so a
|
||||
// locally-accepted item can never poison a whole batch with a server-side 400.
|
||||
const MAX_ITEM_DATA_BYTES = 16_000;
|
||||
// Maximum length of the merged custom parent chain per item; mirrors the server.
|
||||
const MAX_PARENT_CHAIN = 10;
|
||||
// Mirrors the server's UUID_RE — raw parent ids that fail this locally would
|
||||
// 400 the entire batch server-side, so they must never enter the buffer.
|
||||
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;
|
||||
|
||||
/**
|
||||
* Serializable form of a span's identity + full custom ancestor chain. Survives
|
||||
* JSON boundaries (page props, headers), so a span started on one tier can be
|
||||
* continued as a parent on another with full ancestry — unlike a bare uuid
|
||||
* string, which contributes only itself.
|
||||
*/
|
||||
export type SpanRef = {
|
||||
spanId: string,
|
||||
parentSpanIds: string[],
|
||||
};
|
||||
|
||||
/**
|
||||
* Anything accepted as a parent: a raw span uuid (contributes only itself — its
|
||||
* ancestors are unknowable), a serialized SpanRef, or a live Span handle (both
|
||||
* contribute their full ancestor chain plus themselves).
|
||||
*/
|
||||
export type ParentRef = string | SpanRef | Span;
|
||||
|
||||
export type TrackOptions = {
|
||||
parentIds?: ParentRef[],
|
||||
};
|
||||
|
||||
export type StartSpanOptions = {
|
||||
data?: Record<string, unknown>,
|
||||
parentIds?: ParentRef[],
|
||||
startedAtMs?: number,
|
||||
};
|
||||
|
||||
/**
|
||||
* A custom span: a time interval written to analytics as an open interval on
|
||||
* start and re-written (versioned upsert) on setData/end. A span that is never
|
||||
* ended — e.g. the tab closed — stays visible as an open interval by design.
|
||||
*
|
||||
* Returned promises resolve when the batch containing the update is acknowledged
|
||||
* and reject on definitive send failure; they are pre-caught internally, so
|
||||
* ignoring them never causes unhandled-rejection noise. No method throws.
|
||||
*/
|
||||
export type Span = {
|
||||
readonly spanId: string,
|
||||
readonly spanType: string,
|
||||
readonly isEnded: boolean,
|
||||
/** Shallow-merges into the span's data and re-writes the span. */
|
||||
setData(data: Record<string, unknown>): Promise<void>,
|
||||
/** Idempotent; repeated calls return the first call's promise. */
|
||||
end(options?: { endedAtMs?: number }): Promise<void>,
|
||||
/** Tracks an event with this span (and its full ancestor chain) as a parent. */
|
||||
trackEvent(eventType: string, data?: Record<string, unknown>, options?: TrackOptions): Promise<void>,
|
||||
/** Starts a child span of this span. */
|
||||
startSpan(spanType: string, options?: StartSpanOptions): Span,
|
||||
/** Serializable identity + full custom ancestor chain (see SpanRef). */
|
||||
ref(): SpanRef,
|
||||
};
|
||||
|
||||
export type SpanUpdateRow = {
|
||||
span_id: string,
|
||||
span_type: string,
|
||||
started_at_ms: number,
|
||||
ended_at_ms: number | null,
|
||||
parent_span_ids: string[],
|
||||
data: Record<string, unknown>,
|
||||
updated_at_ms: number,
|
||||
};
|
||||
|
||||
type Settler = {
|
||||
resolve: () => void,
|
||||
reject: (error: unknown) => void,
|
||||
};
|
||||
|
||||
// Subscribing an internal no-op handler means an ignored rejection never counts
|
||||
// as "unhandled" (the runtime only reports rejected promises with zero
|
||||
// subscribers), while callers who do await still observe it through their own
|
||||
// handler. This is what makes fire-and-forget usage safe.
|
||||
function preCaught<T>(promise: Promise<T>): Promise<T> {
|
||||
promise.catch(() => {});
|
||||
return promise;
|
||||
}
|
||||
|
||||
export function rejectedPreCaught(message: string): Promise<never> {
|
||||
console.error(`Hexclave analytics: ${message}`);
|
||||
return preCaught(Promise.reject(new Error(message)));
|
||||
}
|
||||
|
||||
let warnedTelemetryUnavailable = false;
|
||||
export function warnTelemetryUnavailableOnce(): void {
|
||||
if (warnedTelemetryUnavailable) return;
|
||||
warnedTelemetryUnavailable = true;
|
||||
console.warn("Hexclave analytics: trackEvent/startSpan called where analytics is unavailable (non-browser environment, no persistent token store, or analytics disabled); telemetry is dropped");
|
||||
}
|
||||
|
||||
export function getCustomTelemetryNameError(kind: "event" | "span", name: unknown): string | null {
|
||||
if (typeof name !== "string" || !CUSTOM_TELEMETRY_NAME_RE.test(name)) {
|
||||
return `Invalid custom ${kind} type ${JSON.stringify(name)}: must start with a letter, contain only letters, digits, "_", ".", ":" or "-", and be at most 64 characters ("$"-prefixed names are reserved for system telemetry)`;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
export function getCustomTelemetryDataError(data: unknown): string | null {
|
||||
if (data === undefined) return null;
|
||||
if (data === null || typeof data !== "object" || Array.isArray(data)) {
|
||||
return "Telemetry data must be a plain JSON-serializable object";
|
||||
}
|
||||
// JSON.stringify's lib type is `string` but it returns undefined at runtime
|
||||
// for some inputs — keep the check honest with an explicit widened type.
|
||||
let serialized: string | undefined;
|
||||
try {
|
||||
serialized = JSON.stringify(data) as string | undefined;
|
||||
} catch {
|
||||
return "Telemetry data must be JSON-serializable (no circular references or BigInt values)";
|
||||
}
|
||||
if (serialized === undefined || serialized.length > MAX_ITEM_DATA_BYTES) {
|
||||
return `Telemetry data must serialize to at most ${MAX_ITEM_DATA_BYTES} bytes`;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Merges ambient parents (e.g. global spans) and explicit ParentRefs into one
|
||||
* root-first, deduped id chain. Each Span/SpanRef contributes its full frozen
|
||||
* ancestor chain plus itself; a raw string uuid contributes only itself.
|
||||
* Root-first order is preserved because every contributor is itself root-first
|
||||
* and first-occurrence dedupe keeps the earliest (root-most) position.
|
||||
*/
|
||||
export function resolveParentIds(opts: {
|
||||
explicit?: ParentRef[],
|
||||
ambient?: SpanRef[],
|
||||
}): { ids: string[] } | { error: string } {
|
||||
const chains: string[][] = [];
|
||||
for (const ambient of opts.ambient ?? []) {
|
||||
chains.push([...ambient.parentSpanIds, ambient.spanId]);
|
||||
}
|
||||
for (const parent of opts.explicit ?? []) {
|
||||
if (typeof parent === "string") {
|
||||
chains.push([parent]);
|
||||
} else {
|
||||
const ref = "ref" in parent && typeof parent.ref === "function" ? parent.ref() : parent as SpanRef;
|
||||
chains.push([...ref.parentSpanIds, ref.spanId]);
|
||||
}
|
||||
}
|
||||
const seen = new Set<string>();
|
||||
const merged: string[] = [];
|
||||
for (const chain of chains) {
|
||||
for (const id of chain) {
|
||||
if (!UUID_RE.test(id)) {
|
||||
return { error: `Invalid parent span id ${JSON.stringify(id)}: parent ids must be span uuids` };
|
||||
}
|
||||
if (!seen.has(id)) {
|
||||
seen.add(id);
|
||||
merged.push(id);
|
||||
}
|
||||
}
|
||||
}
|
||||
if (merged.length > MAX_PARENT_CHAIN) {
|
||||
console.warn(`Hexclave analytics: parent chain exceeds ${MAX_PARENT_CHAIN} spans; keeping the ${MAX_PARENT_CHAIN} nearest ancestors`);
|
||||
return { ids: merged.slice(-MAX_PARENT_CHAIN) };
|
||||
}
|
||||
return { ids: merged };
|
||||
}
|
||||
|
||||
/**
|
||||
* A Span that records nothing. Returned wherever analytics cannot run (SSR,
|
||||
* analytics disabled, tracker torn down) so isomorphic user code never needs to
|
||||
* branch: every method succeeds immediately.
|
||||
*/
|
||||
export function createInertSpan(spanType: string): Span {
|
||||
let ended = false;
|
||||
const span: Span = {
|
||||
spanId: generateUuid(),
|
||||
spanType,
|
||||
get isEnded() {
|
||||
return ended;
|
||||
},
|
||||
setData: () => Promise.resolve(),
|
||||
end: () => {
|
||||
ended = true;
|
||||
return Promise.resolve();
|
||||
},
|
||||
trackEvent: () => Promise.resolve(),
|
||||
startSpan: (childType: string) => createInertSpan(childType),
|
||||
ref: () => ({ spanId: span.spanId, parentSpanIds: [] }),
|
||||
};
|
||||
return span;
|
||||
}
|
||||
|
||||
function hasScreenDimensions(value: unknown): value is { width: number, height: number } {
|
||||
if (value == null || typeof value !== "object") {
|
||||
return false;
|
||||
@ -108,9 +308,14 @@ export type EventTrackerDeps = {
|
||||
};
|
||||
|
||||
type TrackedEvent = {
|
||||
event_type: "$page-view" | "$click",
|
||||
// System types ($page-view, $click) from the auto-capture paths, or a custom
|
||||
// name (validated against CUSTOM_TELEMETRY_NAME_RE) from trackEvent().
|
||||
event_type: string,
|
||||
event_at_ms: number,
|
||||
data: Record<string, unknown>,
|
||||
// Custom ancestor chain, root-first, raw span uuids. Omitted for system
|
||||
// events; the server composes system ancestry on top for every event.
|
||||
parent_span_ids?: string[],
|
||||
};
|
||||
|
||||
export class EventTracker {
|
||||
@ -128,6 +333,22 @@ export class EventTracker {
|
||||
private _originalPushState: History["pushState"] | null = null;
|
||||
private _originalReplaceState: History["replaceState"] | null = null;
|
||||
|
||||
// Custom-span updates awaiting the next flush, latest row per span id (a span
|
||||
// touched N times within one flush window costs one wire row). Settlers from
|
||||
// superseded rows are carried over so every returned promise still settles
|
||||
// with the batch that actually carries the span's latest state.
|
||||
private _spanUpdates = new Map<string, { row: SpanUpdateRow, settlers: Settler[] }>();
|
||||
// Settlers for buffered custom events (system events are fire-and-forget).
|
||||
private _eventSettlers = new Map<TrackedEvent, Settler>();
|
||||
// Spans registered via setGlobalSpan — ambient parents for all subsequent
|
||||
// custom events and spans until unset (end() auto-unsets).
|
||||
private _globalSpans = new Set<Span>();
|
||||
// Live (un-ended) span handles' inert switches; flipped on clearBuffer so a
|
||||
// span started before sign-out can never be re-written under the next user.
|
||||
private _liveSpanControls = new Set<{ markInert: () => void }>();
|
||||
// Batch sends currently on the wire; flush() awaits these.
|
||||
private _inFlight = new Set<Promise<void>>();
|
||||
|
||||
private _deadClickTimer: ReturnType<typeof setInterval> | null = null;
|
||||
private _deadClickMutationObserver: MutationObserver | null = null;
|
||||
// Buffered $click events still awaiting dead-click classification. Always a
|
||||
@ -176,11 +397,35 @@ export class EventTracker {
|
||||
}
|
||||
|
||||
clearBuffer() {
|
||||
this._settleAllPending("analytics buffer cleared");
|
||||
this._events = [];
|
||||
this._approxBytes = 0;
|
||||
this._unclassifiedClicks.clear();
|
||||
}
|
||||
|
||||
// Rejects every pending custom-event/span promise (pre-caught, so silent for
|
||||
// fire-and-forget callers), drops buffered span rows, and inert-ifies all live
|
||||
// span handles. Called on sign-out (paired with the segment-id rotation): a
|
||||
// span started under user A must never be re-written under user B's session.
|
||||
private _settleAllPending(reason: string) {
|
||||
const error = new Error(`Hexclave analytics: ${reason}`);
|
||||
for (const settler of this._eventSettlers.values()) {
|
||||
settler.reject(error);
|
||||
}
|
||||
this._eventSettlers.clear();
|
||||
for (const entry of this._spanUpdates.values()) {
|
||||
for (const settler of entry.settlers) {
|
||||
settler.reject(error);
|
||||
}
|
||||
}
|
||||
this._spanUpdates.clear();
|
||||
for (const control of this._liveSpanControls) {
|
||||
control.markInert();
|
||||
}
|
||||
this._liveSpanControls.clear();
|
||||
this._globalSpans.clear();
|
||||
}
|
||||
|
||||
/**
|
||||
* Replaces the per-tab id shared with the SessionRecorder. Called on sign-out
|
||||
* (paired with clearBuffer) so a subsequent same-tab sign-in as a different user
|
||||
@ -192,13 +437,200 @@ export class EventTracker {
|
||||
this._sessionReplaySegmentId = id;
|
||||
}
|
||||
|
||||
/**
|
||||
* Buffers a custom analytics event. The returned promise resolves when the
|
||||
* batch carrying the event is acknowledged and rejects on definitive send
|
||||
* failure; it is pre-caught, so ignoring it is safe. Never throws — invalid
|
||||
* input yields a rejected promise plus a console error.
|
||||
*/
|
||||
trackCustomEvent(eventType: string, data?: Record<string, unknown>, options?: TrackOptions): Promise<void> {
|
||||
const nameError = getCustomTelemetryNameError("event", eventType);
|
||||
if (nameError) return rejectedPreCaught(nameError);
|
||||
const dataError = getCustomTelemetryDataError(data);
|
||||
if (dataError) return rejectedPreCaught(dataError);
|
||||
const resolved = resolveParentIds({ explicit: options?.parentIds, ambient: this._ambientParentRefs() });
|
||||
if ("error" in resolved) return rejectedPreCaught(resolved.error);
|
||||
if (this._disabled) return Promise.resolve();
|
||||
|
||||
const event: TrackedEvent = {
|
||||
event_type: eventType,
|
||||
event_at_ms: Date.now(),
|
||||
data: { ...data ?? {} },
|
||||
...resolved.ids.length > 0 ? { parent_span_ids: resolved.ids } : {},
|
||||
};
|
||||
let settler!: Settler;
|
||||
const promise = preCaught(new Promise<void>((resolve, reject) => {
|
||||
settler = { resolve, reject };
|
||||
}));
|
||||
this._eventSettlers.set(event, settler);
|
||||
this._pushEvent(event);
|
||||
return promise;
|
||||
}
|
||||
|
||||
/**
|
||||
* Starts a custom span: the open interval is written on the next flush and
|
||||
* re-written (versioned upsert) on setData/end. Never throws — invalid input
|
||||
* yields an inert span plus a console error, so caller code always proceeds.
|
||||
*/
|
||||
startSpan(spanType: string, options?: StartSpanOptions): Span {
|
||||
const nameError = getCustomTelemetryNameError("span", spanType);
|
||||
if (nameError) {
|
||||
console.error(`Hexclave analytics: ${nameError}`);
|
||||
return createInertSpan(spanType);
|
||||
}
|
||||
const dataError = getCustomTelemetryDataError(options?.data);
|
||||
if (dataError) {
|
||||
console.error(`Hexclave analytics: ${dataError}`);
|
||||
return createInertSpan(spanType);
|
||||
}
|
||||
if (options?.startedAtMs !== undefined && (!Number.isInteger(options.startedAtMs) || options.startedAtMs < 0)) {
|
||||
console.error(`Hexclave analytics: startedAtMs must be a non-negative integer epoch-milliseconds value`);
|
||||
return createInertSpan(spanType);
|
||||
}
|
||||
const resolved = resolveParentIds({ explicit: options?.parentIds, ambient: this._ambientParentRefs() });
|
||||
if ("error" in resolved) {
|
||||
console.error(`Hexclave analytics: ${resolved.error}`);
|
||||
return createInertSpan(spanType);
|
||||
}
|
||||
if (this._disabled) return createInertSpan(spanType);
|
||||
|
||||
const spanId = generateUuid();
|
||||
// The custom ancestor chain is frozen at creation: parents are identity, not
|
||||
// state, so later setGlobalSpan calls or parent mutations never re-parent an
|
||||
// existing span (and every re-write of this span carries the same chain).
|
||||
const parentSpanIds = resolved.ids;
|
||||
const startedAtMs = options?.startedAtMs ?? Date.now();
|
||||
let accumulatedData: Record<string, unknown> = { ...options?.data ?? {} };
|
||||
let lastVersion = 0;
|
||||
let ended = false;
|
||||
let inert = false;
|
||||
let endPromise: Promise<void> | null = null;
|
||||
|
||||
// Per-span monotonic version: the row carrying the latest update always wins
|
||||
// in the ReplacingMergeTree, even when batches arrive out of order (keepalive
|
||||
// sends are single-attempt and can race a normal flush).
|
||||
const nextVersion = () => (lastVersion = Math.max(Date.now(), lastVersion + 1));
|
||||
const enqueue = (endedAtMs: number | null): Promise<void> => {
|
||||
if (inert || this._disabled) return Promise.resolve();
|
||||
return this._enqueueSpanUpdate({
|
||||
span_id: spanId,
|
||||
span_type: spanType,
|
||||
started_at_ms: startedAtMs,
|
||||
ended_at_ms: endedAtMs,
|
||||
parent_span_ids: parentSpanIds,
|
||||
data: { ...accumulatedData },
|
||||
updated_at_ms: nextVersion(),
|
||||
});
|
||||
};
|
||||
|
||||
const control = {
|
||||
markInert: () => {
|
||||
inert = true;
|
||||
},
|
||||
};
|
||||
this._liveSpanControls.add(control);
|
||||
|
||||
const span: Span = {
|
||||
spanId,
|
||||
spanType,
|
||||
get isEnded() {
|
||||
return ended;
|
||||
},
|
||||
setData: (data: Record<string, unknown>) => {
|
||||
if (ended) return rejectedPreCaught(`setData() called on already-ended span "${spanType}"`);
|
||||
const merged = { ...accumulatedData, ...data };
|
||||
const mergedError = getCustomTelemetryDataError(merged);
|
||||
if (mergedError) return rejectedPreCaught(mergedError);
|
||||
accumulatedData = merged;
|
||||
return enqueue(null);
|
||||
},
|
||||
end: (endOptions?: { endedAtMs?: number }) => {
|
||||
if (endPromise) return endPromise;
|
||||
ended = true;
|
||||
this._globalSpans.delete(span);
|
||||
this._liveSpanControls.delete(control);
|
||||
// Clamp so a caller-supplied end can never invert the interval — the
|
||||
// server rejects ended < started, and one bad item would 400 the batch.
|
||||
const endedAtMs = Math.max(startedAtMs, Math.round(endOptions?.endedAtMs ?? Date.now()));
|
||||
endPromise = enqueue(endedAtMs);
|
||||
return endPromise;
|
||||
},
|
||||
trackEvent: (eventType: string, data?: Record<string, unknown>, trackOptions?: TrackOptions) =>
|
||||
this.trackCustomEvent(eventType, data, { ...trackOptions, parentIds: [span, ...trackOptions?.parentIds ?? []] }),
|
||||
startSpan: (childType: string, childOptions?: StartSpanOptions) =>
|
||||
this.startSpan(childType, { ...childOptions, parentIds: [span, ...childOptions?.parentIds ?? []] }),
|
||||
ref: () => ({ spanId, parentSpanIds: [...parentSpanIds] }),
|
||||
};
|
||||
|
||||
// Write the open interval right away: a span the user never ends (e.g. the
|
||||
// tab closes) still shows up, as an open interval, from its first flush on.
|
||||
enqueue(null).catch(() => {});
|
||||
return span;
|
||||
}
|
||||
|
||||
/**
|
||||
* Registers a span as an ambient parent for all subsequently created custom
|
||||
* events and spans (additive with explicit parentIds). Ending the span
|
||||
* automatically unregisters it.
|
||||
*/
|
||||
setGlobalSpan(span: Span): void {
|
||||
if (span.isEnded) {
|
||||
console.warn("Hexclave analytics: setGlobalSpan() called with an already-ended span; ignoring");
|
||||
return;
|
||||
}
|
||||
this._globalSpans.add(span);
|
||||
}
|
||||
|
||||
unsetGlobalSpan(span: Span): void {
|
||||
this._globalSpans.delete(span);
|
||||
}
|
||||
|
||||
/**
|
||||
* Sends everything buffered right now and settles all in-flight sends. This is
|
||||
* the "send now" escape hatch — awaiting trackEvent alone waits for the
|
||||
* regular flush cadence.
|
||||
*/
|
||||
async flush(): Promise<void> {
|
||||
await this._flush({ keepalive: false });
|
||||
await Promise.allSettled([...this._inFlight]);
|
||||
}
|
||||
|
||||
private _ambientParentRefs(): SpanRef[] {
|
||||
const refs: SpanRef[] = [];
|
||||
for (const span of this._globalSpans) {
|
||||
if (!span.isEnded) refs.push(span.ref());
|
||||
}
|
||||
return refs;
|
||||
}
|
||||
|
||||
private _enqueueSpanUpdate(row: SpanUpdateRow): Promise<void> {
|
||||
let settler!: Settler;
|
||||
const promise = preCaught(new Promise<void>((resolve, reject) => {
|
||||
settler = { resolve, reject };
|
||||
}));
|
||||
const previous = this._spanUpdates.get(row.span_id);
|
||||
if (previous) {
|
||||
this._approxBytes -= JSON.stringify(previous.row).length;
|
||||
}
|
||||
// Latest row per span id wins within a batch, but superseded rows' settlers
|
||||
// ride along so their promises still settle with the batch that ships.
|
||||
this._spanUpdates.set(row.span_id, { row, settlers: [...previous?.settlers ?? [], settler] });
|
||||
this._approxBytes += JSON.stringify(row).length;
|
||||
this._maybeTriggerSizeFlush();
|
||||
return promise;
|
||||
}
|
||||
|
||||
private _maybeTriggerSizeFlush() {
|
||||
if (this._events.length + this._spanUpdates.size >= MAX_EVENTS_PER_BATCH || this._approxBytes >= MAX_APPROX_BYTES_PER_BATCH) {
|
||||
runAsynchronously(() => this._flush({ keepalive: false }));
|
||||
}
|
||||
}
|
||||
|
||||
private _pushEvent(event: TrackedEvent) {
|
||||
if (this._disabled) return;
|
||||
this._events.push(event);
|
||||
this._approxBytes += JSON.stringify(event).length;
|
||||
if (this._events.length >= MAX_EVENTS_PER_BATCH || this._approxBytes >= MAX_APPROX_BYTES_PER_BATCH) {
|
||||
runAsynchronously(() => this._flush({ keepalive: false }));
|
||||
}
|
||||
this._maybeTriggerSizeFlush();
|
||||
}
|
||||
|
||||
private _capturePageView(entryType: "initial" | "push" | "replace" | "pop") {
|
||||
@ -480,6 +912,7 @@ export class EventTracker {
|
||||
document.removeEventListener("click", this._onClickCapture, { capture: true });
|
||||
this._teardownDeadClickDetection();
|
||||
|
||||
this._settleAllPending("analytics tracker stopped");
|
||||
this._events = [];
|
||||
this._approxBytes = 0;
|
||||
}
|
||||
@ -496,12 +929,29 @@ export class EventTracker {
|
||||
|
||||
// Clicks still awaiting classification stay buffered so the sweep can
|
||||
// mark them dead in place; classification finishes well within one flush
|
||||
// interval, so they ride the next flush at the latest.
|
||||
// interval, so they ride the next flush at the latest. Span rows are never
|
||||
// held back — the holdback exists only for dead-click classification.
|
||||
const events = this._events.filter((event) => !this._unclassifiedClicks.has(event));
|
||||
if (events.length === 0) return;
|
||||
const spanEntries = [...this._spanUpdates.values()];
|
||||
if (events.length === 0 && spanEntries.length === 0) return;
|
||||
this._events = this._events.filter((event) => this._unclassifiedClicks.has(event));
|
||||
this._spanUpdates.clear();
|
||||
this._approxBytes = this._events.reduce((total, event) => total + JSON.stringify(event).length, 0);
|
||||
|
||||
// Snapshot the settlers of everything this batch carries: they settle with
|
||||
// this send's outcome. Items buffered after this point ride the next flush.
|
||||
const settlers: Settler[] = [];
|
||||
for (const event of events) {
|
||||
const settler = this._eventSettlers.get(event);
|
||||
if (settler) {
|
||||
settlers.push(settler);
|
||||
this._eventSettlers.delete(event);
|
||||
}
|
||||
}
|
||||
for (const entry of spanEntries) {
|
||||
settlers.push(...entry.settlers);
|
||||
}
|
||||
|
||||
const nowMs = Date.now();
|
||||
|
||||
const batchId = generateUuid();
|
||||
@ -510,30 +960,54 @@ export class EventTracker {
|
||||
batch_id: batchId,
|
||||
sent_at_ms: nowMs,
|
||||
events,
|
||||
...spanEntries.length > 0 ? { spans: spanEntries.map((entry) => entry.row) } : {},
|
||||
};
|
||||
|
||||
const res = await this._deps.sendBatch(
|
||||
JSON.stringify(payload),
|
||||
{ keepalive: options.keepalive },
|
||||
);
|
||||
const send = (async () => {
|
||||
try {
|
||||
const res = await this._deps.sendBatch(
|
||||
JSON.stringify(payload),
|
||||
{ keepalive: options.keepalive },
|
||||
);
|
||||
|
||||
if (res.status === "error") {
|
||||
if (isAnalyticsNotEnabledError(res.error)) {
|
||||
this._disable();
|
||||
return;
|
||||
}
|
||||
// Ad blockers commonly block analytics endpoints, causing network
|
||||
// errors. These are expected and should not pollute the console.
|
||||
if (isAdBlockerNetworkError(res.error)) {
|
||||
return;
|
||||
}
|
||||
console.warn("EventTracker flush failed:", res.error);
|
||||
return;
|
||||
}
|
||||
if (res.status === "error") {
|
||||
// All rejections are pre-caught at promise creation, so failures are
|
||||
// silent for fire-and-forget callers and observable for awaiting ones.
|
||||
for (const settler of settlers) settler.reject(res.error);
|
||||
if (isAnalyticsNotEnabledError(res.error)) {
|
||||
this._disable();
|
||||
return;
|
||||
}
|
||||
// Ad blockers commonly block analytics endpoints, causing network
|
||||
// errors. These are expected and should not pollute the console.
|
||||
if (isAdBlockerNetworkError(res.error)) {
|
||||
return;
|
||||
}
|
||||
console.warn("EventTracker flush failed:", res.error);
|
||||
return;
|
||||
}
|
||||
|
||||
if (!res.data.ok) {
|
||||
console.warn("EventTracker flush failed:", res.data.status, await res.data.text());
|
||||
}
|
||||
if (!res.data.ok) {
|
||||
const text = await res.data.text();
|
||||
for (const settler of settlers) settler.reject(new Error(`EventTracker flush failed: ${res.data.status} ${text}`));
|
||||
console.warn("EventTracker flush failed:", res.data.status, text);
|
||||
return;
|
||||
}
|
||||
|
||||
for (const settler of settlers) settler.resolve();
|
||||
} catch (error) {
|
||||
// _flush must never reject (public flush() and fire-and-forget callers
|
||||
// don't expect telemetry failures to throw); the settlers carry it.
|
||||
for (const settler of settlers) settler.reject(error);
|
||||
console.warn("EventTracker flush failed:", error);
|
||||
}
|
||||
})();
|
||||
|
||||
const tracked: Promise<void> = send.finally(() => {
|
||||
this._inFlight.delete(tracked);
|
||||
});
|
||||
this._inFlight.add(tracked);
|
||||
await tracked;
|
||||
}
|
||||
|
||||
private _disable() {
|
||||
@ -547,7 +1021,7 @@ export class EventTracker {
|
||||
|
||||
private _tick() {
|
||||
if (this._cancelled) return;
|
||||
if (this._events.length > 0) {
|
||||
if (this._events.length > 0 || this._spanUpdates.size > 0) {
|
||||
runAsynchronously(() => this._flush({ keepalive: false }));
|
||||
}
|
||||
}
|
||||
|
||||
@ -36,6 +36,8 @@ import { ProjectCurrentServerUser, ServerOAuthProvider, ServerUser, ServerUserCr
|
||||
import { StackServerAppConstructorOptions } from "../interfaces/server-app";
|
||||
import { _HexclaveClientAppImplIncomplete } from "./client-app-impl";
|
||||
import { clientVersion, createCache, createCacheBySession, getDefaultExtraRequestHeaders, getDefaultProjectId, getDefaultPublishableClientKey, getDefaultSecretServerKey, resolveApiUrls, resolveConstructorOptions } from "./common";
|
||||
import { createInertSpan, getCustomTelemetryDataError, getCustomTelemetryNameError, rejectedPreCaught, resolveParentIds, type Span, type SpanRef, type SpanUpdateRow, type StartSpanOptions, type TrackOptions } from "./event-tracker";
|
||||
import { generateUuid } from "./session-replay";
|
||||
|
||||
import { useAsyncCache } from "./common"; // THIS_LINE_PLATFORM react-like
|
||||
|
||||
@ -1698,4 +1700,294 @@ export class _HexclaveServerAppImplIncomplete<HasTokenStore extends boolean, Pro
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Custom telemetry (server-key sends)
|
||||
//
|
||||
// The browser tracker (EventTracker) only exists in browser-like environments;
|
||||
// on the server there is no session to derive identity from, so trackEvent/
|
||||
// startSpan take an explicit `userId` and authenticate with the secret server
|
||||
// key. Items coalesce per (userId) in a buffer flushed on the next microtask —
|
||||
// a synchronous loop of N trackEvent calls costs one POST, not N — while every
|
||||
// call still gets its own settled-on-ack promise. `await` (or flush()) is the
|
||||
// delivery guarantee: there is no page-lifetime flush cadence on the server.
|
||||
//
|
||||
// NOTE: setGlobalSpan is app-instance-level state. Under concurrent requests
|
||||
// (one shared app instance) a global span set in one request becomes a parent
|
||||
// in all of them — prefer explicit parentIds (or span.trackEvent) on servers.
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
private readonly _serverTelemetryBuffers = new Map<string | null, ServerTelemetryBuffer>();
|
||||
private readonly _serverGlobalSpans = new Set<Span>();
|
||||
private readonly _serverTelemetryInFlight = new Set<Promise<void>>();
|
||||
|
||||
override trackEvent(eventType: string, data?: Record<string, unknown>, options?: TrackOptions & { userId?: string }): Promise<void> {
|
||||
if (this._eventTracker) {
|
||||
// Browser-like environment: identity comes from the session; an explicit
|
||||
// userId would silently mis-attribute, so refuse it loudly.
|
||||
if (options?.userId !== undefined) {
|
||||
return rejectedPreCaught("userId is only supported for server-key telemetry; in the browser, events are attributed to the signed-in user");
|
||||
}
|
||||
return this._eventTracker.trackCustomEvent(eventType, data, options);
|
||||
}
|
||||
return this._trackServerEvent(eventType, data, options, options?.userId ?? null);
|
||||
}
|
||||
|
||||
override startSpan(spanType: string, options?: StartSpanOptions & { userId?: string }): Span {
|
||||
if (this._eventTracker) {
|
||||
if (options?.userId !== undefined) {
|
||||
console.error("Hexclave analytics: userId is only supported for server-key telemetry; in the browser, spans are attributed to the signed-in user");
|
||||
return createInertSpan(spanType);
|
||||
}
|
||||
return this._eventTracker.startSpan(spanType, options);
|
||||
}
|
||||
return this._startServerSpan(spanType, options, options?.userId ?? null);
|
||||
}
|
||||
|
||||
override setGlobalSpan(span: Span): void {
|
||||
if (this._eventTracker) {
|
||||
this._eventTracker.setGlobalSpan(span);
|
||||
return;
|
||||
}
|
||||
if (span.isEnded) {
|
||||
console.warn("Hexclave analytics: setGlobalSpan() called with an already-ended span; ignoring");
|
||||
return;
|
||||
}
|
||||
this._serverGlobalSpans.add(span);
|
||||
}
|
||||
|
||||
override unsetGlobalSpan(span: Span): void {
|
||||
this._eventTracker?.unsetGlobalSpan(span);
|
||||
this._serverGlobalSpans.delete(span);
|
||||
}
|
||||
|
||||
override async flush(): Promise<void> {
|
||||
await super.flush();
|
||||
for (const key of [...this._serverTelemetryBuffers.keys()]) {
|
||||
this._flushServerTelemetry(key);
|
||||
}
|
||||
await Promise.allSettled([...this._serverTelemetryInFlight]);
|
||||
}
|
||||
|
||||
private _serverAmbientParentRefs(): SpanRef[] {
|
||||
const refs: SpanRef[] = [];
|
||||
for (const span of this._serverGlobalSpans) {
|
||||
if (!span.isEnded) refs.push(span.ref());
|
||||
}
|
||||
return refs;
|
||||
}
|
||||
|
||||
private _trackServerEvent(eventType: string, data: Record<string, unknown> | undefined, options: TrackOptions | undefined, userId: string | null): Promise<void> {
|
||||
const nameError = getCustomTelemetryNameError("event", eventType);
|
||||
if (nameError) return rejectedPreCaught(nameError);
|
||||
const dataError = getCustomTelemetryDataError(data);
|
||||
if (dataError) return rejectedPreCaught(dataError);
|
||||
if (userId !== null && !SERVER_TELEMETRY_UUID_RE.test(userId)) {
|
||||
return rejectedPreCaught(`Invalid userId ${JSON.stringify(userId)}: must be a user uuid`);
|
||||
}
|
||||
const resolved = resolveParentIds({ explicit: options?.parentIds, ambient: this._serverAmbientParentRefs() });
|
||||
if ("error" in resolved) return rejectedPreCaught(resolved.error);
|
||||
|
||||
let settler!: TelemetrySettler;
|
||||
const promise = new Promise<void>((resolve, reject) => {
|
||||
settler = { resolve, reject };
|
||||
});
|
||||
promise.catch(() => {});
|
||||
const buffer = this._getServerTelemetryBuffer(userId);
|
||||
buffer.events.push({
|
||||
event: {
|
||||
event_type: eventType,
|
||||
event_at_ms: Date.now(),
|
||||
data: { ...data ?? {} },
|
||||
...resolved.ids.length > 0 ? { parent_span_ids: resolved.ids } : {},
|
||||
},
|
||||
settler,
|
||||
});
|
||||
this._afterServerTelemetryEnqueue(userId, buffer);
|
||||
return promise;
|
||||
}
|
||||
|
||||
private _startServerSpan(spanType: string, options: StartSpanOptions | undefined, userId: string | null): Span {
|
||||
const nameError = getCustomTelemetryNameError("span", spanType);
|
||||
if (nameError) {
|
||||
console.error(`Hexclave analytics: ${nameError}`);
|
||||
return createInertSpan(spanType);
|
||||
}
|
||||
const dataError = getCustomTelemetryDataError(options?.data);
|
||||
if (dataError) {
|
||||
console.error(`Hexclave analytics: ${dataError}`);
|
||||
return createInertSpan(spanType);
|
||||
}
|
||||
if (options?.startedAtMs !== undefined && (!Number.isInteger(options.startedAtMs) || options.startedAtMs < 0)) {
|
||||
console.error("Hexclave analytics: startedAtMs must be a non-negative integer epoch-milliseconds value");
|
||||
return createInertSpan(spanType);
|
||||
}
|
||||
if (userId !== null && !SERVER_TELEMETRY_UUID_RE.test(userId)) {
|
||||
console.error(`Hexclave analytics: invalid userId ${JSON.stringify(userId)}: must be a user uuid`);
|
||||
return createInertSpan(spanType);
|
||||
}
|
||||
const resolved = resolveParentIds({ explicit: options?.parentIds, ambient: this._serverAmbientParentRefs() });
|
||||
if ("error" in resolved) {
|
||||
console.error(`Hexclave analytics: ${resolved.error}`);
|
||||
return createInertSpan(spanType);
|
||||
}
|
||||
|
||||
const spanId = generateUuid();
|
||||
const parentSpanIds = resolved.ids;
|
||||
const startedAtMs = options?.startedAtMs ?? Date.now();
|
||||
let accumulatedData: Record<string, unknown> = { ...options?.data ?? {} };
|
||||
let lastVersion = 0;
|
||||
let ended = false;
|
||||
let endPromise: Promise<void> | null = null;
|
||||
|
||||
const nextVersion = () => (lastVersion = Math.max(Date.now(), lastVersion + 1));
|
||||
const enqueue = (endedAtMs: number | null): Promise<void> => this._enqueueServerSpanUpdate(userId, {
|
||||
span_id: spanId,
|
||||
span_type: spanType,
|
||||
started_at_ms: startedAtMs,
|
||||
ended_at_ms: endedAtMs,
|
||||
parent_span_ids: parentSpanIds,
|
||||
data: { ...accumulatedData },
|
||||
updated_at_ms: nextVersion(),
|
||||
});
|
||||
|
||||
const span: Span = {
|
||||
spanId,
|
||||
spanType,
|
||||
get isEnded() {
|
||||
return ended;
|
||||
},
|
||||
setData: (data: Record<string, unknown>) => {
|
||||
if (ended) return rejectedPreCaught(`setData() called on already-ended span "${spanType}"`);
|
||||
const merged = { ...accumulatedData, ...data };
|
||||
const mergedError = getCustomTelemetryDataError(merged);
|
||||
if (mergedError) return rejectedPreCaught(mergedError);
|
||||
accumulatedData = merged;
|
||||
return enqueue(null);
|
||||
},
|
||||
end: (endOptions?: { endedAtMs?: number }) => {
|
||||
if (endPromise) return endPromise;
|
||||
ended = true;
|
||||
this._serverGlobalSpans.delete(span);
|
||||
const endedAtMs = Math.max(startedAtMs, Math.round(endOptions?.endedAtMs ?? Date.now()));
|
||||
endPromise = enqueue(endedAtMs);
|
||||
return endPromise;
|
||||
},
|
||||
trackEvent: (eventType: string, data?: Record<string, unknown>, trackOptions?: TrackOptions) =>
|
||||
this._trackServerEvent(eventType, data, { ...trackOptions, parentIds: [span, ...trackOptions?.parentIds ?? []] }, userId),
|
||||
startSpan: (childType: string, childOptions?: StartSpanOptions) =>
|
||||
this._startServerSpan(childType, { ...childOptions, parentIds: [span, ...childOptions?.parentIds ?? []] }, userId),
|
||||
ref: () => ({ spanId, parentSpanIds: [...parentSpanIds] }),
|
||||
};
|
||||
|
||||
enqueue(null).catch(() => {});
|
||||
return span;
|
||||
}
|
||||
|
||||
private _enqueueServerSpanUpdate(userId: string | null, row: SpanUpdateRow): Promise<void> {
|
||||
let settler!: TelemetrySettler;
|
||||
const promise = new Promise<void>((resolve, reject) => {
|
||||
settler = { resolve, reject };
|
||||
});
|
||||
promise.catch(() => {});
|
||||
const buffer = this._getServerTelemetryBuffer(userId);
|
||||
const previous = buffer.spans.get(row.span_id);
|
||||
// Latest row per span id wins within a batch; superseded rows' settlers ride
|
||||
// along so every returned promise settles with the batch that ships.
|
||||
buffer.spans.set(row.span_id, { row, settlers: [...previous?.settlers ?? [], settler] });
|
||||
this._afterServerTelemetryEnqueue(userId, buffer);
|
||||
return promise;
|
||||
}
|
||||
|
||||
private _getServerTelemetryBuffer(userId: string | null): ServerTelemetryBuffer {
|
||||
let buffer = this._serverTelemetryBuffers.get(userId);
|
||||
if (!buffer) {
|
||||
buffer = { events: [], spans: new Map(), scheduled: false };
|
||||
this._serverTelemetryBuffers.set(userId, buffer);
|
||||
}
|
||||
return buffer;
|
||||
}
|
||||
|
||||
private _afterServerTelemetryEnqueue(userId: string | null, buffer: ServerTelemetryBuffer): void {
|
||||
// Stay well under the server's 500-items-per-batch cap: ship immediately once
|
||||
// the coalesced batch is large, otherwise wait for the microtask boundary.
|
||||
if (buffer.events.length + buffer.spans.size >= SERVER_TELEMETRY_MAX_ITEMS_PER_BATCH) {
|
||||
this._flushServerTelemetry(userId);
|
||||
return;
|
||||
}
|
||||
if (buffer.scheduled) return;
|
||||
buffer.scheduled = true;
|
||||
queueMicrotask(() => this._flushServerTelemetry(userId));
|
||||
}
|
||||
|
||||
private _flushServerTelemetry(userId: string | null): void {
|
||||
const buffer = this._serverTelemetryBuffers.get(userId);
|
||||
if (!buffer) return;
|
||||
this._serverTelemetryBuffers.delete(userId);
|
||||
const events = buffer.events;
|
||||
const spanEntries = [...buffer.spans.values()];
|
||||
if (events.length === 0 && spanEntries.length === 0) return;
|
||||
|
||||
const settlers: TelemetrySettler[] = events.map((entry) => entry.settler);
|
||||
for (const entry of spanEntries) {
|
||||
settlers.push(...entry.settlers);
|
||||
}
|
||||
|
||||
const payload = {
|
||||
batch_id: generateUuid(),
|
||||
sent_at_ms: Date.now(),
|
||||
...userId !== null ? { user_id: userId } : {},
|
||||
...events.length > 0 ? { events: events.map((entry) => entry.event) } : {},
|
||||
...spanEntries.length > 0 ? { spans: spanEntries.map((entry) => entry.row) } : {},
|
||||
};
|
||||
|
||||
const send = (async () => {
|
||||
try {
|
||||
const res = await this._interface.sendAnalyticsEventBatchAsServer(JSON.stringify(payload));
|
||||
if (res.status === "error") {
|
||||
for (const settler of settlers) settler.reject(res.error);
|
||||
console.warn("Hexclave analytics: server telemetry send failed:", res.error);
|
||||
return;
|
||||
}
|
||||
if (!res.data.ok) {
|
||||
const text = await res.data.text();
|
||||
for (const settler of settlers) settler.reject(new Error(`Hexclave analytics: server telemetry send failed: ${res.data.status} ${text}`));
|
||||
console.warn("Hexclave analytics: server telemetry send failed:", res.data.status, text);
|
||||
return;
|
||||
}
|
||||
for (const settler of settlers) settler.resolve();
|
||||
} catch (error) {
|
||||
for (const settler of settlers) settler.reject(error);
|
||||
console.warn("Hexclave analytics: server telemetry send failed:", error);
|
||||
}
|
||||
})();
|
||||
const tracked: Promise<void> = send.finally(() => {
|
||||
this._serverTelemetryInFlight.delete(tracked);
|
||||
});
|
||||
this._serverTelemetryInFlight.add(tracked);
|
||||
}
|
||||
}
|
||||
|
||||
const SERVER_TELEMETRY_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;
|
||||
// Below the route's 500-items cap with headroom, so a coalesced batch can never
|
||||
// be rejected for size.
|
||||
const SERVER_TELEMETRY_MAX_ITEMS_PER_BATCH = 400;
|
||||
|
||||
type TelemetrySettler = {
|
||||
resolve: () => void,
|
||||
reject: (error: unknown) => void,
|
||||
};
|
||||
|
||||
type ServerTelemetryBuffer = {
|
||||
events: {
|
||||
event: {
|
||||
event_type: string,
|
||||
event_at_ms: number,
|
||||
data: Record<string, unknown>,
|
||||
parent_span_ids?: string[],
|
||||
},
|
||||
settler: TelemetrySettler,
|
||||
}[],
|
||||
spans: Map<string, { row: SpanUpdateRow, settlers: TelemetrySettler[] }>,
|
||||
scheduled: boolean,
|
||||
};
|
||||
|
||||
@ -7,6 +7,7 @@ import { CustomerInvoicesList, CustomerInvoicesRequestOptions, CustomerProductsL
|
||||
import { Project } from "../../projects";
|
||||
import { ProjectCurrentUser, SyncedPartialUser, TokenPartialUser } from "../../users";
|
||||
import { _HexclaveClientAppImpl } from "../implementations";
|
||||
import type { Span, StartSpanOptions, TrackOptions } from "../implementations/event-tracker";
|
||||
import { AnalyticsOptions } from "../implementations/session-replay";
|
||||
|
||||
/** @deprecated Use `HexclaveClientAppConstructorOptions` from the `@hexclave/*` package instead — same symbol, new brand name. See https://docs.hexclave.com/migration. */
|
||||
@ -112,6 +113,38 @@ export type StackClientApp<HasTokenStore extends boolean = boolean, ProjectId ex
|
||||
|
||||
cancelSubscription(options: { productId: string, subscriptionId?: string } | { productId: string, subscriptionId?: string, teamId: string }): Promise<void>,
|
||||
|
||||
/**
|
||||
* Tracks a custom analytics event. Buffered and sent in batches. The
|
||||
* returned promise resolves when the batch carrying the event is
|
||||
* acknowledged (up to one flush interval later — call `flush()` to send
|
||||
* immediately) and is safe to ignore. Never throws; invalid input yields a
|
||||
* rejected (pre-caught) promise. No-ops outside the browser or when
|
||||
* analytics is disabled.
|
||||
*/
|
||||
trackEvent(eventType: string, data?: Record<string, unknown>, options?: TrackOptions): Promise<void>,
|
||||
|
||||
/**
|
||||
* Starts a custom span (a time interval). The span is written immediately
|
||||
* as an open interval and re-written when data changes or it ends — a span
|
||||
* that is never ended (e.g. the tab closed) stays visible as an open
|
||||
* interval. Never throws; returns an inert no-op span outside the browser,
|
||||
* when analytics is disabled, or on invalid input.
|
||||
*/
|
||||
startSpan(spanType: string, options?: StartSpanOptions): Span,
|
||||
|
||||
/**
|
||||
* Registers a span as an ambient parent for all subsequently tracked
|
||||
* custom events and spans (additive with explicit `parentIds`). Ending the
|
||||
* span automatically unregisters it.
|
||||
*/
|
||||
setGlobalSpan(span: Span): void,
|
||||
unsetGlobalSpan(span: Span): void,
|
||||
|
||||
/**
|
||||
* Sends all buffered analytics immediately and settles in-flight sends.
|
||||
*/
|
||||
flush(): Promise<void>,
|
||||
|
||||
// note: we don't special-case 'anonymous' here to return non-null, see GetPartialUserOptions for more details
|
||||
getPartialUser(options: GetCurrentPartialUserOptions<HasTokenStore> & { from: 'token' }): Promise<TokenPartialUser | null>,
|
||||
getPartialUser(options: GetCurrentPartialUserOptions<HasTokenStore> & { from: 'convex' }): Promise<TokenPartialUser | null>,
|
||||
|
||||
@ -8,6 +8,7 @@ import { EmailDeliveryInfo, SendEmailOptions } from "../../email";
|
||||
import { ServerListTeamsOptions, ServerListUsersOptions, ServerTeam, ServerTeamCreateOptions } from "../../teams";
|
||||
import { ProjectCurrentServerUser, ServerOAuthProvider, ServerUser, ServerUserCreateOptions, SyncedPartialServerUser, TokenPartialUser } from "../../users";
|
||||
import { _HexclaveServerAppImpl } from "../implementations";
|
||||
import type { Span, StartSpanOptions, TrackOptions } from "../implementations/event-tracker";
|
||||
import { StackClientApp, StackClientAppConstructorOptions } from "./client-app";
|
||||
|
||||
|
||||
@ -37,6 +38,20 @@ export type StackServerApp<HasTokenStore extends boolean = boolean, ProjectId ex
|
||||
{ returnUrl?: string }
|
||||
)): Promise<string>,
|
||||
|
||||
/**
|
||||
* Server-side variant of `trackEvent`: attribution is explicit via `userId`
|
||||
* (there is no session to derive it from). Items coalesce per userId and
|
||||
* send on the next microtask; `await` the promise (or call `flush()`) as the
|
||||
* delivery guarantee — the server has no page-lifetime flush cadence.
|
||||
*/
|
||||
trackEvent(eventType: string, data?: Record<string, unknown>, options?: TrackOptions & { userId?: string }): Promise<void>,
|
||||
|
||||
/**
|
||||
* Server-side variant of `startSpan`: attribution is explicit via `userId`.
|
||||
* Child spans and span-attached events inherit the span's userId.
|
||||
*/
|
||||
startSpan(spanType: string, options?: StartSpanOptions & { userId?: string }): Span,
|
||||
|
||||
// IF_PLATFORM react-like
|
||||
useUser(options: GetCurrentUserOptions<HasTokenStore> & { or: 'redirect' }): ProjectCurrentServerUser<ProjectId>,
|
||||
useUser(options: GetCurrentUserOptions<HasTokenStore> & { or: 'throw' }): ProjectCurrentServerUser<ProjectId>,
|
||||
|
||||
Loading…
Reference in New Issue
Block a user