From e0c8dedb4b1f31981a7f222dc27ce8243e633237 Mon Sep 17 00:00:00 2001 From: David Condrey Date: Tue, 4 Aug 2026 15:50:13 -0700 Subject: [PATCH] feat(server): read the Analytics Engine mirror behind a breakdown endpoint --- apps/server/src/db/breakdown.ts | 234 +++++++++++++++++ apps/server/src/lib/ae-sql.ts | 132 ++++++++++ apps/server/src/lib/ae.ts | 25 +- apps/server/src/routes/llms.ts | 4 + apps/server/src/routes/stats.ts | 21 ++ apps/server/test/ae-read.test.ts | 428 +++++++++++++++++++++++++++++++ docs/api.md | 37 +++ docs/self-hosting.md | 18 ++ packages/shared/src/schemas.ts | 49 ++++ packages/shared/src/stats.ts | 34 ++- 10 files changed, 979 insertions(+), 3 deletions(-) create mode 100644 apps/server/src/db/breakdown.ts create mode 100644 apps/server/src/lib/ae-sql.ts create mode 100644 apps/server/test/ae-read.test.ts diff --git a/apps/server/src/db/breakdown.ts b/apps/server/src/db/breakdown.ts new file mode 100644 index 0000000..414379e --- /dev/null +++ b/apps/server/src/db/breakdown.ts @@ -0,0 +1,234 @@ +// Single-dimension breakdowns, served from the columnar store when this deployment has one and from +// D1 otherwise. Both paths answer the SAME question over the same events; the response says which +// one answered, because only one of them samples. +// +// This is the first read that goes to Analytics Engine. It is a NEW endpoint rather than a swap +// underneath an existing one on purpose: every figure on `/api/stats` is exact today, and quietly +// re-sourcing it would trade that for scale nobody asked for. What this adds is the dimensions D1 +// has always stored but no endpoint ever surfaced — city, timezone, the three UTM columns, form +// factor, currency, hostname — with the ordinary filters composed on top. + +import type { + BreakdownDimension, + BreakdownResponse, + BreakdownRow, + StatsFilter, +} from '@facet/shared'; +import { desc, sql } from 'drizzle-orm'; +import type { SQLiteColumn } from 'drizzle-orm/sqlite-core'; +import type { Env } from '../env.js'; +import { type FetchLike, aeInt, aeLiteral, aeNumber, queryAe } from '../lib/ae-sql.js'; +import { AE_DATASET, type MirroredColumn, VISITOR_BLOB, blobColumn } from '../lib/ae.js'; +import { db } from './queries.js'; +import * as schema from './schema.js'; +import { K_ANON, buildFilteredEventWhere, pageviewCount } from './stats.js'; + +/** Each groupable dimension's two addresses: the mirrored column it occupies in the columnar store, + * and the `events` column it occupies in D1. Declared as a total `Record`, so adding a dimension to + * `BREAKDOWN_DIMENSIONS` without teaching BOTH stores about it is a type error rather than a + * runtime hole. */ +const DIMENSIONS: Record = { + hostname: { blob: 'hostname', column: schema.events.hostname }, + path: { blob: 'path', column: schema.events.path }, + referrer: { blob: 'referrer', column: schema.events.referrer }, + event: { blob: 'name', column: schema.events.name }, + country: { blob: 'country', column: schema.events.country }, + region: { blob: 'region', column: schema.events.region }, + city: { blob: 'city', column: schema.events.city }, + timezone: { blob: 'timezone', column: schema.events.timezone }, + network: { blob: 'network', column: schema.events.network }, + language: { blob: 'language', column: schema.events.language }, + device: { blob: 'device', column: schema.events.device }, + form_factor: { blob: 'formFactor', column: schema.events.formFactor }, + browser: { blob: 'browser', column: schema.events.browser }, + os: { blob: 'os', column: schema.events.os }, + channel: { blob: 'channel', column: schema.events.channel }, + utm_source: { blob: 'utmSource', column: schema.events.utmSource }, + utm_medium: { blob: 'utmMedium', column: schema.events.utmMedium }, + utm_campaign: { blob: 'utmCampaign', column: schema.events.utmCampaign }, + currency: { blob: 'currency', column: schema.events.currency }, +}; + +/** The exact-match filters `buildFilteredEventWhere` applies in D1, paired with the mirrored column + * each one narrows in the columnar store, so the two stores filter on the same values. `hostname` is + * absent because D1 treats it as truthy-or-absent rather than defined-or-absent — see `aeWhere`. */ +const FILTERS = [ + ['path', 'path'], + ['referrer', 'referrer'], + ['country', 'country'], + ['device', 'device'], + ['channel', 'channel'], +] as const satisfies readonly (readonly [keyof StatsFilter, MirroredColumn])[]; + +/** The `WHERE` terms for a filter, or `null` when the columnar store cannot express one of them and + * must decline the whole read. Dropping the offending term instead would return UNFILTERED rows + * under a filtered label, which is the failure `/api/stats/distribution` already refuses to ship. */ +function aeWhere(f: StatsFilter): string[] | null { + const site = aeLiteral(f.siteId); + // Data points carry a second-granular `timestamp`, so a sub-second range boundary is widened to + // the second enclosing it. Every range this endpoint serves is minute-aligned or coarser. + const start = aeInt(Math.floor(f.start / 1000)); + const end = aeInt(Math.ceil(f.end / 1000)); + if (site === null || start === null || end === null) { + return null; + } + const terms = [ + `index1 = ${site}`, + `toUInt32(timestamp) >= ${start}`, + `toUInt32(timestamp) < ${end}`, + ]; + // Mirrors D1's truthiness test: an empty hostname is no filter at all there, so it is none here. + if (f.hostname) { + const literal = aeLiteral(f.hostname); + if (literal === null) { + return null; + } + terms.push(`${blobColumn('hostname')} = ${literal}`); + } + for (const [key, blob] of FILTERS) { + const value = f[key]; + if (typeof value !== 'string') { + continue; + } + // The one value the two stores disagree on. D1 keeps an absent dimension as NULL (and an + // absent referrer as ''), while the columnar store has no NULL and keeps both as '' — so + // `country=''` matches nothing in D1 and every country-less row here. Decline and let D1 + // answer, rather than return a different result set under the same query string. + const literal = value === '' ? null : aeLiteral(value); + if (literal === null) { + return null; + } + terms.push(`${blobColumn(blob)} = ${literal}`); + } + return terms; +} + +/** + * The breakdown query, or `null` when it cannot be built safely. + * + * Every count is weighted by `_sample_interval`: the columnar store samples under load, and a bare + * `count()` reports the surviving rows rather than the traffic they stand for. `visitors` is the + * exception that cannot be corrected — a distinct count of sampled rows is a lower bound, no weight + * recovers the identities that were dropped — which is why the response carries `sampled`. + */ +function aeBreakdownSql( + f: StatsFilter, + dimension: BreakdownDimension, + limit: number, +): string | null { + const where = aeWhere(f); + const rows = aeInt(limit); + if (where === null || rows === null) { + return null; + } + const key = blobColumn(DIMENSIONS[dimension].blob); + return [ + `SELECT ${key} AS k,`, + 'SUM(_sample_interval) AS total,', + 'SUM(_sample_interval * double3) AS pageviews,', + `count(DISTINCT ${VISITOR_BLOB}) AS visitors,`, + 'max(_sample_interval) AS sample_interval', + `FROM ${AE_DATASET}`, + `WHERE ${where.join(' AND ')}`, + 'GROUP BY k', + // The same k-anonymity floor D1 applies, on DISTINCT VISITORS rather than events: a group of + // three pageviews by one person is one person, and this endpoint reaches dimensions (city, + // campaign) where that distinction is the whole risk. Sampling only tightens it. + `HAVING visitors >= ${K_ANON}`, + 'ORDER BY total DESC, k ASC', + `LIMIT ${rows}`, + 'FORMAT JSON', + ].join(' '); +} + +interface AeBreakdownRow { + k?: unknown; + total?: unknown; + pageviews?: unknown; + visitors?: unknown; + sample_interval?: unknown; +} + +/** Read the breakdown from the columnar store, or `null` when this deployment cannot serve it. */ +async function aeBreakdown( + env: Env, + f: StatsFilter, + dimension: BreakdownDimension, + limit: number, + fetchImpl?: FetchLike, +): Promise { + const query = aeBreakdownSql(f, dimension, limit); + if (query === null) { + return null; + } + const data = await queryAe(env, query, fetchImpl); + if (data === null) { + return null; + } + return { + dimension, + source: 'analytics_engine', + sampled: data.some((r) => aeNumber(r.sample_interval) > 1), + rows: data.map((r) => ({ + key: typeof r.k === 'string' ? r.k : String(r.k ?? ''), + // Sampling weights are per-row multipliers, so a weighted sum is fractional. Report whole + // events: a breakdown claiming 41.7 pageviews is an estimate advertised as a measurement. + events: Math.round(aeNumber(r.total)), + pageviews: Math.round(aeNumber(r.pageviews)), + visitors: Math.round(aeNumber(r.visitors)), + })), + }; +} + +/** Read the breakdown from D1 — always exact, and the answer whenever the columnar store declines. */ +async function d1Breakdown( + env: Env, + f: StatsFilter, + dimension: BreakdownDimension, + limit: number, +): Promise { + // Fold NULL to '' so an absent dimension carries the same key it does in the columnar store, + // which has no NULL. Without this the two sources would label the same group differently. + const key = sql`COALESCE(${DIMENSIONS[dimension].column}, '')`; + const total = sql`COUNT(*)`; + const visitors = sql`COUNT(DISTINCT ${schema.events.visitorHash})`; + const rows = await db(env) + .select({ key, total, pageviews: pageviewCount, visitors }) + .from(schema.events) + .where(buildFilteredEventWhere(f)) + .groupBy(key) + .having(sql`${visitors} >= ${K_ANON}`) + .orderBy(desc(total), key) + .limit(limit); + return rows.map((r) => ({ + key: String(r.key ?? ''), + events: Number(r.total ?? 0), + pageviews: Number(r.pageviews ?? 0), + visitors: Number(r.visitors ?? 0), + })); +} + +/** + * Group the range by one dimension. Prefers the columnar store and falls back to D1 for every reason + * it can decline — unbound, unconfigured, retention-gated, a filter value it cannot express safely, + * or a query the API rejected. The fallback is not a degraded mode: D1 holds the same events and + * answers exactly, it just scans to do it. + */ +export async function breakdown( + env: Env, + f: StatsFilter, + dimension: BreakdownDimension, + limit: number, + fetchImpl?: FetchLike, +): Promise { + const columnar = await aeBreakdown(env, f, dimension, limit, fetchImpl); + if (columnar !== null) { + return columnar; + } + return { + dimension, + source: 'd1', + sampled: false, + rows: await d1Breakdown(env, f, dimension, limit), + }; +} diff --git a/apps/server/src/lib/ae-sql.ts b/apps/server/src/lib/ae-sql.ts new file mode 100644 index 0000000..3761f5a --- /dev/null +++ b/apps/server/src/lib/ae-sql.ts @@ -0,0 +1,132 @@ +// Analytics Engine reads: SQL over HTTP against the dataset `lib/ae.ts` mirrors every accepted event +// into. This is the read half of the columnar store — D1 stays authoritative and answers every +// existing endpoint, and a caller that cannot be served here falls back to it rather than degrading. +// +// THE INJECTION BOUNDARY LIVES HERE. Analytics Engine SQL has no bound parameters: a query is a +// string of text posted to an account-scoped endpoint under a bearer token, so every value a caller +// influences is concatenated into that text or into that URL. Two rules follow, and every builder in +// the codebase goes through them: +// +// 1. A column is never a caller string. Dimensions are resolved through `blobColumn` against the +// `BLOB_SCHEMA` key set, so the only reachable columns are the twenty the write path fills. +// 2. A value is only interpolated when it is provably inexpressible as syntax. The SQL reference +// documents string literals as single-quoted and documents NO escape sequence for a quote or a +// backslash inside one, so there is no escaping rule to implement correctly — `aeLiteral` +// therefore REFUSES a value containing either, and the caller falls back to D1, which answers +// it exactly through a bound parameter. Guessing at an undocumented escape would be the one +// place in this codebase where a query's meaning depends on an unverified vendor behaviour. + +import type { Env } from '../env.js'; +import { AE_RETENTION_DAYS } from './ae.js'; +import { createLogger } from './log.js'; +import { retentionDays } from './retention.js'; + +/** Bound on a single AE read, so a slow or hung analytics API cannot hold a dashboard request open. */ +const AE_QUERY_TIMEOUT_MS = 10_000; + +/** Cloudflare account ids are 32 lowercase hex characters. This value is interpolated into the API + * URL PATH, and the request that follows carries `CF_API_TOKEN` — so an id containing `/`, `@`, or a + * `..` segment could rewrite the request target and hand the token to a host of the operator's + * typo's choosing. Validated to the exact shape rather than merely non-empty. */ +const ACCOUNT_ID = /^[0-9a-f]{32}$/; + +/** Rejects any value that could end a `'…'` literal or smuggle syntax into one: a single quote, a + * backslash (the escape character in every dialect this parser could be built on), or a control + * character. What remains cannot terminate the literal, so it cannot be read as SQL. */ +const UNSAFE_IN_LITERAL = /['\\\p{Cc}]/u; + +/** The subset of `fetch` this module uses, injectable so tests drive the client without a network. */ +export type FetchLike = (url: string, init: RequestInit) => Promise; + +/** + * Whether this deployment can read from Analytics Engine at all. + * + * The binding is required even though a read does not use it: with `AE` unbound nothing was ever + * mirrored, so a query would report an empty dataset as an empty site. The retention gate is the + * same one `writeEvent` applies, and for the same reason — below `AE_RETENTION_DAYS` the write path + * declines, so the dataset holds nothing to read. + */ +export function aeReadable(env: Env): boolean { + return ( + env.AE !== undefined && + ACCOUNT_ID.test(env.CF_ACCOUNT_ID ?? '') && + (env.CF_API_TOKEN ?? '') !== '' && + retentionDays(env) >= AE_RETENTION_DAYS + ); +} + +/** A quoted string literal, or `null` when the value cannot be expressed as one safely. Callers MUST + * treat `null` as "this store cannot answer the query" — never as "drop the filter", which would + * return unfiltered rows under a filtered label. */ +export function aeLiteral(value: string): string | null { + return UNSAFE_IN_LITERAL.test(value) ? null : `'${value}'`; +} + +/** An integer literal, or `null` for anything that is not an exact integer (NaN, Infinity, 1e21 — + * each of which `String()` would happily render into the query as something else). */ +export function aeInt(value: number): string | null { + return Number.isSafeInteger(value) ? String(value) : null; +} + +/** The `FORMAT JSON` envelope. `meta`/`rows` are ignored: `data` is the only field a caller reads, + * and every cell is re-validated by the caller because ClickHouse renders wide integers as strings. */ +interface AeSqlResponse { + data?: unknown; +} + +/** + * Run one query and return its rows, or `null` when the read could not be completed for ANY reason — + * unconfigured, rejected, timed out, or malformed. There is no error to propagate because there is + * nothing for a caller to do differently: D1 holds the same events and answers exactly. + */ +export async function queryAe( + env: Env, + sql: string, + fetchImpl: FetchLike = fetch as unknown as FetchLike, +): Promise { + if (!aeReadable(env)) { + return null; + } + const log = createLogger({ component: 'ae-sql' }); + const controller = new AbortController(); + const timer = setTimeout(() => controller.abort(), AE_QUERY_TIMEOUT_MS); + try { + const res = await fetchImpl( + `https://api.cloudflare.com/client/v4/accounts/${env.CF_ACCOUNT_ID}/analytics_engine/sql`, + { + method: 'POST', + headers: { + authorization: `Bearer ${env.CF_API_TOKEN}`, + 'content-type': 'text/plain; charset=utf-8', + }, + body: sql, + signal: controller.signal, + }, + ); + if (!res.ok) { + // Status only. The response body of a rejected analytics query echoes the query text, and + // this deployment's logs are not where a site's paths and referrers should surface. + log.warn('ae_query_rejected', { status: res.status }); + return null; + } + const body = (await res.json()) as AeSqlResponse; + if (!Array.isArray(body.data)) { + log.warn('ae_query_malformed'); + return null; + } + return body.data as T[]; + } catch (err) { + log.error('ae_query_failed', err); + return null; + } finally { + clearTimeout(timer); + } +} + +/** Coerce one AE cell to a finite number. Aggregates come back as JSON numbers for doubles but as + * decimal STRINGS for 64-bit integer sums, so a bare `as number` silently yields `"12"` and every + * arithmetic on it becomes string concatenation. */ +export function aeNumber(cell: unknown): number { + const n = typeof cell === 'string' || typeof cell === 'number' ? Number(cell) : Number.NaN; + return Number.isFinite(n) ? n : 0; +} diff --git a/apps/server/src/lib/ae.ts b/apps/server/src/lib/ae.ts index ce88ed5..d0175df 100644 --- a/apps/server/src/lib/ae.ts +++ b/apps/server/src/lib/ae.ts @@ -37,7 +37,7 @@ type StringColumn = { * fixed handful of enum values that D1 groups from an index in milliseconds, so mirroring them buys * nothing and spends a slot a genuinely high-cardinality dimension may need later. */ -const BLOB_SCHEMA: readonly { readonly key: StringColumn; readonly bytes: number }[] = [ +const BLOB_SCHEMA = [ { key: 'hostname', bytes: 253 }, { key: 'path', bytes: 1024 }, { key: 'referrer', bytes: 1024 }, @@ -58,7 +58,28 @@ const BLOB_SCHEMA: readonly { readonly key: StringColumn; readonly bytes: number { key: 'utmMedium', bytes: 200 }, { key: 'utmCampaign', bytes: 200 }, { key: 'currency', bytes: 3 }, -]; +] as const satisfies readonly { key: StringColumn; bytes: number }[]; + +/** A column this deployment actually mirrors — the key set of `BLOB_SCHEMA`, narrowed to literals so + * a read can only ever name a slot the write path fills. */ +export type MirroredColumn = (typeof BLOB_SCHEMA)[number]['key']; + +/** The `blobN` column a mirrored key occupies, 1-based, derived from `BLOB_SCHEMA` itself. Reads go + * through this so the layout has exactly ONE definition: appending a slot cannot leave a query + * addressing the position the column used to hold. */ +export function blobColumn(key: MirroredColumn): string { + return `blob${BLOB_SCHEMA.findIndex((slot) => slot.key === key) + 1}`; +} + +/** The `blobN` column carrying the derived visitor hash. Reads reference it ONLY inside + * `count(DISTINCT …)`: it is the one mirrored column that identifies a browsing session rather than + * describing it, so it must never become a group key or a projected value. */ +export const VISITOR_BLOB = blobColumn('visitorHash'); + +/** The dataset name reads query in their `FROM` clause. The binding object exposes no name at + * runtime, so this MUST stay equal to `analytics_engine_datasets[].dataset` in `wrangler.jsonc`; a + * rename there without a change here queries a table that was never written to. */ +export const AE_DATASET = 'facet_events'; /** Analytics Engine per-data-point limits: one index of at most 96 bytes, at most 20 blobs totalling * at most 16 KB, at most 20 doubles. Exported so the tests assert the layout stays inside them diff --git a/apps/server/src/routes/llms.ts b/apps/server/src/routes/llms.ts index d0c1c88..d572b1a 100644 --- a/apps/server/src/routes/llms.ts +++ b/apps/server/src/routes/llms.ts @@ -39,6 +39,10 @@ All read endpoints take an API key issued by this deployment's operator, sent as - \`GET /api/stats?site_id&start&end&interval=hour|day\` — full stats document, plus optional \`path\`, \`referrer\`, \`country\`, \`device\`, \`channel\` filters. +- \`GET /api/stats/breakdown?site_id&start&end&dimension=&limit=\` — group the range by one + dimension, including the ones no other endpoint surfaces (\`city\`, \`timezone\`, \`utm_source\`, + \`utm_medium\`, \`utm_campaign\`, \`form_factor\`, \`currency\`). Groups below 3 distinct visitors + are withheld; read \`sampled\` before quoting the figures. - \`GET /api/stats/realtime?site_id\` — active visitors in a trailing 5-minute window. - \`GET /api/stats/sessions?site_id&start&end\` — sessions and engagement. - \`GET /api/stats/channels?site_id&start&end\` — traffic channel breakdown. diff --git a/apps/server/src/routes/stats.ts b/apps/server/src/routes/stats.ts index 63b8c92..7ed83ac 100644 --- a/apps/server/src/routes/stats.ts +++ b/apps/server/src/routes/stats.ts @@ -2,6 +2,8 @@ // owns the requested site, and assembles the full stats response. import { + BREAKDOWN_DEFAULT_ROWS, + BreakdownQuerySchema, type CountRow, DimensionSeriesQuerySchema, type Goal, @@ -22,6 +24,7 @@ import { vValidator } from '@hono/valibot-validator'; import { eq } from 'drizzle-orm'; import { Hono } from 'hono'; import { detectAnomalies } from '../db/anomaly.js'; +import { breakdown } from '../db/breakdown.js'; import { listExperiments, listFunnels, listGoals } from '../db/catalog.js'; import { goalConversions } from '../db/conversions.js'; import { experimentResult } from '../db/experiments.js'; @@ -256,6 +259,24 @@ statsRoutes.get( }, ); +// One dimension, grouped over the range, with the ordinary filters composed on top. This is the read +// that uses the columnar mirror: it reaches the dimensions D1 stores but no other endpoint surfaces +// (city, timezone, the UTM columns, form factor, currency, hostname), and it falls back to D1 +// whenever the mirror is absent or cannot express the query. `source`/`sampled` in the response say +// which store answered, because only the columnar one samples — see `BreakdownResponse`. +statsRoutes.get( + '/stats/breakdown', + requireSiteAccess, + vValidator('query', BreakdownQuerySchema, validationErrorHook), + async (c) => { + const query = c.req.valid('query'); + const f = toStatsFilter(query, c.get('siteId')); + return c.json( + await breakdown(c.env, f, query.dimension, query.limit ?? BREAKDOWN_DEFAULT_ROWS), + ); + }, +); + // Internal/system interactions ($exposure, form_submit, other $-prefixed) shown separately from // marketer-facing custom events, which exclude them. statsRoutes.get( diff --git a/apps/server/test/ae-read.test.ts b/apps/server/test/ae-read.test.ts new file mode 100644 index 0000000..a2e9008 --- /dev/null +++ b/apps/server/test/ae-read.test.ts @@ -0,0 +1,428 @@ +// The Analytics Engine read path. The subject under test is mostly the SAFETY of a query the vendor +// gives us no bound parameters to build: this proves the literal guard cannot be talked past, that +// the account id reaching the URL — under a bearer token — is validated to its exact shape, that a +// value the guard rejects falls back to D1 instead of silently losing its filter, and that the two +// stores answer the same question. Sampling correction is asserted on the emitted SQL, because no +// local runtime samples. + +import { env } from 'cloudflare:test'; +import type { BreakdownDimension, StatsFilter } from '@facet/shared'; +import { beforeEach, describe, expect, it, vi } from 'vitest'; +import { createApp } from '../src/app.js'; +import { breakdown } from '../src/db/breakdown.js'; +import { type NewEvent, insertEvent } from '../src/db/queries.js'; +import type { Env } from '../src/env.js'; +import { aeInt, aeLiteral, aeNumber, aeReadable, queryAe } from '../src/lib/ae-sql.js'; +import { AE_RETENTION_DAYS, VISITOR_BLOB, blobColumn } from '../src/lib/ae.js'; + +const SITE = '77777777-7777-4777-8777-777777777777'; +const ACCOUNT = '0123456789abcdef0123456789abcdef'; +const T0 = Date.UTC(2026, 3, 1); +const HOUR = 3_600_000; + +/** A fully configured read env. `AE` only has to be present — reads go over HTTP, not the binding. */ +function readableEnv(over: Partial = {}): Env { + return { + AE: { writeDataPoint: () => {} }, + CF_ACCOUNT_ID: ACCOUNT, + CF_API_TOKEN: 'test-cf-token', + RAW_RETENTION_DAYS: String(AE_RETENTION_DAYS), + ...over, + } as unknown as Env; +} + +/** A stub `fetch` returning one AE `FORMAT JSON` envelope, recording the request it was given. */ +function stubAe(data: unknown[], init: ResponseInit = {}) { + return vi.fn(async (url: string, req: RequestInit) => { + void url; + void req; + return new Response(JSON.stringify({ meta: [], data, rows: data.length }), { + status: 200, + ...init, + }); + }); +} + +function mk(over: Partial = {}): NewEvent { + return { + siteId: SITE, + hostname: 'shop.example.com', + path: '/', + referrer: '', + name: null, + props: null, + visitorHash: 'v0', + country: 'US', + device: 'desktop', + createdAt: T0, + ...over, + }; +} + +describe('aeLiteral', () => { + it('quotes a value that cannot express syntax', () => { + expect(aeLiteral('/pricing')).toBe("'/pricing'"); + expect(aeLiteral('')).toBe("''"); + // Non-ASCII is not a syntax risk: it cannot terminate a literal. + expect(aeLiteral('München')).toBe("'München'"); + }); + + it('refuses a quote or a backslash rather than guessing an undocumented escape', () => { + expect(aeLiteral("' OR 1=1 --")).toBeNull(); + expect(aeLiteral("o'brien")).toBeNull(); + expect(aeLiteral('C:\\Windows')).toBeNull(); + expect(aeLiteral("\\' union select")).toBeNull(); + }); + + it('refuses control characters, so a value can never carry a line into the query', () => { + expect(aeLiteral('a\nb')).toBeNull(); + expect(aeLiteral('a\r\nb')).toBeNull(); + expect(aeLiteral('a\u0000b')).toBeNull(); + expect(aeLiteral('a\u007fb')).toBeNull(); + }); +}); + +describe('aeInt', () => { + it('renders an exact integer', () => { + expect(aeInt(0)).toBe('0'); + expect(aeInt(1_775_000_000)).toBe('1775000000'); + expect(aeInt(-5)).toBe('-5'); + }); + + it('refuses anything String() would render as something other than digits', () => { + // 1e21 stringifies to "1e+21", NaN/Infinity to words — each of which changes the query. + expect(aeInt(1e21)).toBeNull(); + expect(aeInt(Number.NaN)).toBeNull(); + expect(aeInt(Number.POSITIVE_INFINITY)).toBeNull(); + expect(aeInt(1.5)).toBeNull(); + }); +}); + +describe('aeNumber', () => { + it('reads a 64-bit sum returned as a decimal string', () => { + expect(aeNumber('120')).toBe(120); + expect(aeNumber(4.5)).toBe(4.5); + }); + + it('reads anything unusable as zero rather than NaN', () => { + expect(aeNumber(undefined)).toBe(0); + expect(aeNumber(null)).toBe(0); + expect(aeNumber('not a number')).toBe(0); + expect(aeNumber({})).toBe(0); + }); +}); + +describe('aeReadable', () => { + it('is true only for a fully configured deployment', () => { + expect(aeReadable(readableEnv())).toBe(true); + }); + + it('is false without the dataset binding, since nothing was ever mirrored', () => { + expect(aeReadable(readableEnv({ AE: undefined }))).toBe(false); + }); + + it('is false without a token', () => { + expect(aeReadable(readableEnv({ CF_API_TOKEN: '' }))).toBe(false); + expect(aeReadable(readableEnv({ CF_API_TOKEN: undefined as unknown as string }))).toBe( + false, + ); + }); + + it('refuses an account id that is not exactly 32 lowercase hex', () => { + // Each of these lands in the URL path of a request that carries CF_API_TOKEN. + for (const bad of [ + '', + 'not-an-account', + `${ACCOUNT}/../../../../foo`, + `${ACCOUNT}@evil.example`, + ACCOUNT.toUpperCase(), + ACCOUNT.slice(0, 31), + `${ACCOUNT}0`, + ]) { + expect(aeReadable(readableEnv({ CF_ACCOUNT_ID: bad }))).toBe(false); + } + }); + + it('is false below the window the write path mirrors at, because nothing was written', () => { + expect(aeReadable(readableEnv({ RAW_RETENTION_DAYS: String(AE_RETENTION_DAYS - 1) }))).toBe( + false, + ); + expect(aeReadable(readableEnv({ RAW_RETENTION_DAYS: String(AE_RETENTION_DAYS) }))).toBe( + true, + ); + }); +}); + +describe('queryAe', () => { + it('posts the query text under a bearer token to the account SQL endpoint', async () => { + const fetchImpl = stubAe([{ k: 'x' }]); + const rows = await queryAe(readableEnv(), 'SELECT 1 FORMAT JSON', fetchImpl); + expect(rows).toEqual([{ k: 'x' }]); + const [url, req] = fetchImpl.mock.calls[0] ?? []; + expect(url).toBe( + `https://api.cloudflare.com/client/v4/accounts/${ACCOUNT}/analytics_engine/sql`, + ); + expect(req?.method).toBe('POST'); + expect(req?.body).toBe('SELECT 1 FORMAT JSON'); + expect((req?.headers as Record).authorization).toBe('Bearer test-cf-token'); + }); + + it('never issues the request at all when the deployment is not configured to read', async () => { + const fetchImpl = stubAe([{ k: 'x' }]); + expect(await queryAe({} as Env, 'SELECT 1', fetchImpl)).toBeNull(); + expect(fetchImpl).not.toHaveBeenCalled(); + }); + + it('returns null for a rejected, malformed, or failed response', async () => { + const readable = readableEnv(); + expect(await queryAe(readable, 'SELECT 1', stubAe([], { status: 403 }))).toBeNull(); + expect(await queryAe(readable, 'SELECT 1', stubAe([], { status: 500 }))).toBeNull(); + const notJson = vi.fn(async () => new Response('gateway', { status: 200 })); + expect(await queryAe(readable, 'SELECT 1', notJson)).toBeNull(); + const noData = vi.fn(async () => new Response(JSON.stringify({ error: 'nope' }))); + expect(await queryAe(readable, 'SELECT 1', noData)).toBeNull(); + const threw = vi.fn(async () => { + throw new Error('network'); + }); + expect(await queryAe(readable, 'SELECT 1', threw)).toBeNull(); + }); +}); + +describe('blob layout addressing', () => { + it('maps a mirrored key to its 1-based position', () => { + expect(blobColumn('hostname')).toBe('blob1'); + expect(blobColumn('path')).toBe('blob2'); + expect(blobColumn('visitorHash')).toBe('blob5'); + expect(blobColumn('currency')).toBe('blob20'); + expect(VISITOR_BLOB).toBe('blob5'); + }); +}); + +describe('breakdown → Analytics Engine SQL', () => { + const f: StatsFilter = { siteId: SITE, start: T0, end: T0 + 24 * HOUR }; + + async function sqlFor( + filter: StatsFilter = f, + dimension: BreakdownDimension = 'city', + limit = 25, + ): Promise { + const fetchImpl = stubAe([]); + await breakdown(readableEnv(), filter, dimension, limit, fetchImpl); + return String(fetchImpl.mock.calls[0]?.[1]?.body ?? ''); + } + + it('groups by the allowlisted blob for the dimension, never by a caller string', async () => { + expect(await sqlFor(f, 'city')).toContain(`SELECT ${blobColumn('city')} AS k`); + expect(await sqlFor(f, 'utm_campaign')).toContain( + `SELECT ${blobColumn('utmCampaign')} AS k`, + ); + expect(await sqlFor(f, 'event')).toContain(`SELECT ${blobColumn('name')} AS k`); + }); + + it('weights every count by the sampling interval', async () => { + const sql = await sqlFor(); + expect(sql).toContain('SUM(_sample_interval) AS total'); + expect(sql).toContain('SUM(_sample_interval * double3) AS pageviews'); + expect(sql).toContain('max(_sample_interval) AS sample_interval'); + // A bare count() would report surviving rows as if they were the traffic. + expect(sql).not.toMatch(/\bcount\(\)/); + }); + + it('reads the visitor column only inside a distinct count, never as a group key', async () => { + const sql = await sqlFor(); + expect(sql).toContain(`count(DISTINCT ${VISITOR_BLOB}) AS visitors`); + expect(sql).not.toContain(`${VISITOR_BLOB} AS k`); + expect(sql).not.toContain(`GROUP BY ${VISITOR_BLOB}`); + }); + + it('applies the k-anonymity floor on distinct visitors, and bounds the rows', async () => { + const sql = await sqlFor(f, 'city', 7); + expect(sql).toContain('HAVING visitors >= 3'); + expect(sql).toContain('LIMIT 7'); + }); + + it('scopes to the site and the range in whole seconds', async () => { + const sql = await sqlFor(); + expect(sql).toContain(`index1 = '${SITE}'`); + expect(sql).toContain(`toUInt32(timestamp) >= ${T0 / 1000}`); + expect(sql).toContain(`toUInt32(timestamp) < ${(T0 + 24 * HOUR) / 1000}`); + }); + + it('narrows on each filter through its mirrored column', async () => { + const sql = await sqlFor({ + ...f, + hostname: 'shop.example.com', + path: '/checkout', + country: 'GB', + }); + expect(sql).toContain(`${blobColumn('hostname')} = 'shop.example.com'`); + expect(sql).toContain(`${blobColumn('path')} = '/checkout'`); + expect(sql).toContain(`${blobColumn('country')} = 'GB'`); + }); +}); + +describe('breakdown → fallback to D1', () => { + const f = { siteId: SITE, start: T0, end: T0 + 24 * HOUR }; + + beforeEach(async () => { + await env.DB.prepare('DELETE FROM events').run(); + // Berlin clears the 3-visitor floor; Paris does not, and must not be surfaced. + for (const [i, city] of ['Berlin', 'Berlin', 'Berlin', 'Paris', 'Paris'].entries()) { + await insertEvent(env, mk({ city, visitorHash: `v${i}`, createdAt: T0 + i })); + } + }); + + it('answers from D1 when the deployment is not configured for columnar reads', async () => { + const result = await breakdown(env as Env, f, 'city', 25); + expect(result.source).toBe('d1'); + expect(result.sampled).toBe(false); + expect(result.rows).toEqual([{ key: 'Berlin', events: 3, pageviews: 3, visitors: 3 }]); + }); + + it('suppresses a group below the k-anonymity floor', async () => { + const result = await breakdown(env as Env, f, 'city', 25); + expect(result.rows.map((r) => r.key)).not.toContain('Paris'); + }); + + it('reports an absent dimension as the empty string, matching the columnar store', async () => { + const result = await breakdown(env as Env, f, 'timezone', 25); + expect(result.rows).toEqual([{ key: '', events: 5, pageviews: 5, visitors: 5 }]); + }); + + it('falls back rather than dropping a filter it cannot express safely', async () => { + const readable = { ...(env as unknown as Env), ...readableEnv() } as Env; + for (const path of ["/o'brien", 'C:\\temp', '/a\nb']) { + const fetchImpl = stubAe([]); + const result = await breakdown(readable, { ...f, path }, 'city', 25, fetchImpl); + expect(result.source).toBe('d1'); + // The point: no query was built at all, so no filter could have gone missing from one. + expect(fetchImpl).not.toHaveBeenCalled(); + } + }); + + it('falls back on an empty filter value, the one value the two stores read differently', async () => { + // D1 keeps an absent country as NULL (matching nothing), the columnar store as '' (matching + // every country-less row). Answering from either under the same query string would differ. + const fetchImpl = stubAe([]); + const readable = { ...(env as unknown as Env), ...readableEnv() } as Env; + const result = await breakdown(readable, { ...f, country: '' }, 'city', 25, fetchImpl); + expect(result.source).toBe('d1'); + expect(fetchImpl).not.toHaveBeenCalled(); + }); + + it('falls back when the analytics API rejects the query', async () => { + const readable = { ...(env as unknown as Env), ...readableEnv() } as Env; + const result = await breakdown(readable, f, 'city', 25, stubAe([], { status: 500 })); + expect(result.source).toBe('d1'); + expect(result.rows).toEqual([{ key: 'Berlin', events: 3, pageviews: 3, visitors: 3 }]); + }); +}); + +describe('breakdown → columnar rows', () => { + const f = { siteId: SITE, start: T0, end: T0 + 24 * HOUR }; + + it('reports whole events from weighted sums, and flags a sampled read', async () => { + const fetchImpl = stubAe([ + { k: 'Berlin', total: '120', pageviews: '99.6', visitors: '40', sample_interval: '4' }, + { k: '', total: 8, pageviews: 8, visitors: 8, sample_interval: 1 }, + ]); + const result = await breakdown(readableEnv(), f, 'city', 25, fetchImpl); + expect(result.source).toBe('analytics_engine'); + expect(result.sampled).toBe(true); + expect(result.rows).toEqual([ + { key: 'Berlin', events: 120, pageviews: 100, visitors: 40 }, + { key: '', events: 8, pageviews: 8, visitors: 8 }, + ]); + }); + + it('is not flagged as sampled when every group was read whole', async () => { + const fetchImpl = stubAe([ + { k: 'Berlin', total: 3, pageviews: 3, visitors: 3, sample_interval: 1 }, + ]); + const result = await breakdown(readableEnv(), f, 'city', 25, fetchImpl); + expect(result.sampled).toBe(false); + }); +}); + +describe('GET /api/stats/breakdown', () => { + const app = createApp(); + const ADMIN = 'Bearer test-admin-token'; + let siteId = ''; + let key = ''; + + beforeEach(async () => { + const siteRes = await app.request( + '/api/sites', + { + method: 'POST', + headers: { Authorization: ADMIN, 'content-type': 'application/json' }, + body: JSON.stringify({ name: 'Acme', domain: 'acme.com' }), + }, + env, + ); + siteId = ((await siteRes.json()) as { site: { id: string } }).site.id; + const keyRes = await app.request( + '/api/keys', + { + method: 'POST', + headers: { Authorization: ADMIN, 'content-type': 'application/json' }, + body: JSON.stringify({ site_id: siteId }), + }, + env, + ); + key = ((await keyRes.json()) as { key: string }).key; + await env.DB.prepare('DELETE FROM events').run(); + for (const [i, city] of ['Berlin', 'Berlin', 'Berlin', 'Paris'].entries()) { + await insertEvent(env, mk({ siteId, city, visitorHash: `k${i}`, createdAt: T0 + i })); + } + }); + + function get(query: string, bearer = key) { + return app.request( + `/api/stats/breakdown?site_id=${siteId}&start=${T0}&end=${T0 + 24 * HOUR}&${query}`, + { headers: { Authorization: `Bearer ${bearer}` } }, + env, + ); + } + + it('returns the k-anonymised breakdown with the store that answered', async () => { + const res = await get('dimension=city'); + expect(res.status).toBe(200); + expect(await res.json()).toEqual({ + dimension: 'city', + source: 'd1', + sampled: false, + rows: [{ key: 'Berlin', events: 3, pageviews: 3, visitors: 3 }], + }); + }); + + it('reaches a dimension no other endpoint surfaces', async () => { + for (const dimension of ['utm_campaign', 'timezone', 'currency', 'form_factor']) { + expect((await get(`dimension=${dimension}`)).status).toBe(200); + } + }); + + it('rejects a dimension outside the allowlist rather than reaching for a column', async () => { + for (const dimension of ['visitor_hash', 'props', 'blob5', '']) { + const res = await get(`dimension=${encodeURIComponent(dimension)}`); + expect(res.status).toBe(400); + expect((await res.json()) as { error: string }).toMatchObject({ + error: 'validation_failed', + }); + } + }); + + it('requires a dimension, rather than guessing which question was asked', async () => { + expect((await get('limit=5')).status).toBe(400); + }); + + it('bounds the row count', async () => { + expect((await get('dimension=city&limit=0')).status).toBe(400); + expect((await get('dimension=city&limit=201')).status).toBe(400); + expect((await get('dimension=city&limit=200')).status).toBe(200); + }); + + it('is site-scoped by the API key like every other read', async () => { + expect((await get('dimension=city', 'clk_not-a-real-key')).status).toBe(401); + }); +}); diff --git a/docs/api.md b/docs/api.md index 662f32c..c13e14d 100644 --- a/docs/api.md +++ b/docs/api.md @@ -243,6 +243,43 @@ complete — every event lands in a cell and the totals still reconcile with `GE --- +### `GET /api/stats/breakdown?site_id&start&end&dimension=&limit=` (API key) + +Group the range by **one** dimension. This is the only read that reaches the columns Facet stores +but no other endpoint surfaces — `city`, `timezone`, `utm_source`, `utm_medium`, `utm_campaign`, +`form_factor`, `currency`, `hostname` — and it accepts the same `path` / `referrer` / `country` / +`device` / `channel` filters as `GET /api/stats`. + +`dimension` is required and must be one of: `hostname`, `path`, `referrer`, `event`, `country`, +`region`, `city`, `timezone`, `network`, `language`, `device`, `form_factor`, `browser`, `os`, +`channel`, `utm_source`, `utm_medium`, `utm_campaign`, `currency`. Anything else is +`400 validation_failed`. `limit` is `1..200` (default 25). + +Every group must clear a **k-anonymity floor of 3 distinct visitors** to appear at all, so a +breakdown can never resolve to one person's browsing. An absent dimension value is reported as the +empty string, never as `null`. `dimension=event` is the raw `name` column, so it includes the +internal `$`-prefixed events and `form_submit` that `top_events` on `GET /api/stats` filters out — +use `top_events` / `GET /api/stats/interactions` when you want that split. + +```json +{ + "dimension": "city", + "source": "d1", + "sampled": false, + "rows": [{ "key": "Berlin", "events": 412, "pageviews": 380, "visitors": 96 }] +} +``` + +`source` says which store answered. A deployment with Analytics Engine configured +(`analytics_engine_datasets` bound, plus the `CF_ACCOUNT_ID` var and `CF_API_TOKEN` secret) is +served from the columnar mirror; every other deployment — and any query the mirror cannot express — +falls back to D1, which is always exact. **When `sampled` is `true` the figures are estimates:** +Analytics Engine samples under load, `events` and `pageviews` are sampling-corrected, and `visitors` +is a distinct count that no sampling weight can correct, so it is a lower bound. `source: "d1"` is +always exact and always `"sampled": false`. + +--- + ## Visualization reads Five shapes the cube and the flat top-N lists cannot express: a session distribution diff --git a/docs/self-hosting.md b/docs/self-hosting.md index adc593d..5c1448b 100644 --- a/docs/self-hosting.md +++ b/docs/self-hosting.md @@ -290,6 +290,24 @@ Setting this **below 90 disables the Analytics Engine mirror**. Cloudflare keeps three months with no delete API, so a mirrored copy cannot be purged on your schedule — a shorter window means the deployment stays D1-only rather than retaining data past what it advertises. +## Analytics Engine reads (optional) + +The `analytics_engine_datasets` binding in `apps/server/wrangler.jsonc` is enough to **write** the +columnar mirror. Reading it back — which is what `GET /api/stats/breakdown` uses — goes over +Cloudflare's SQL API rather than the binding, so it additionally needs: + +```sh +# Var: your 32-hex Cloudflare account id. +npx wrangler deploy --var CF_ACCOUNT_ID: +# Secret: an API token with "Account | Account Analytics | Read", and nothing else. +npx wrangler secret put CF_API_TOKEN +``` + +Leave either unset and every read falls back to D1, which answers the same questions exactly — the +mirror is a scale option, not a dependency. Keep the `dataset` name in `wrangler.jsonc` as +`facet_events`: the binding does not expose its own name at runtime, so reads query that name +directly and a rename silently sends every breakdown back to D1. + ## Operations ### Diagnosing an install diff --git a/packages/shared/src/schemas.ts b/packages/shared/src/schemas.ts index 0e9de03..b34a590 100644 --- a/packages/shared/src/schemas.ts +++ b/packages/shared/src/schemas.ts @@ -159,6 +159,54 @@ export const DimensionSeriesQuerySchema = v.object({ limit: v.optional(boundedIntParam(1, SERIES_MAX_KEYS)), }); +/** + * Every dimension `GET /api/stats/breakdown` can group by — one per column the columnar mirror + * carries, so the endpoint answers identically whichever store serves it. + * + * `visitor_hash` is mirrored too and is deliberately NOT here: it is the one column that identifies + * a browsing session rather than describing it, and grouping by it would return one row per person. + * The four enum-shaped columns D1 keeps but does not mirror (screen tier, connection, orientation, + * DPR class) are absent for the opposite reason — `/api/stats` already breaks them down. + */ +export const BREAKDOWN_DIMENSIONS = [ + 'hostname', + 'path', + 'referrer', + 'event', + 'country', + 'region', + 'city', + 'timezone', + 'network', + 'language', + 'device', + 'form_factor', + 'browser', + 'os', + 'channel', + 'utm_source', + 'utm_medium', + 'utm_campaign', + 'currency', +] as const; + +export type BreakdownDimension = (typeof BREAKDOWN_DIMENSIONS)[number]; + +/** Upper bound on breakdown rows. Higher than the fixed top-N lists on `/api/stats` because the + * point of this endpoint is the long tail, but still a constant rather than the data's cardinality. */ +export const BREAKDOWN_MAX_ROWS = 200; + +/** Default breakdown depth when the caller does not ask for one. */ +export const BREAKDOWN_DEFAULT_ROWS = 25; + +/** Query for `GET /api/stats/breakdown`: the stats query plus which dimension to group by. The same + * `path`/`referrer`/`country`/`device`/`channel` filters narrow it as they do everywhere else. */ +export const BreakdownQuerySchema = v.object({ + ...StatsQuerySchema.entries, + dimension: v.picklist(BREAKDOWN_DIMENSIONS), + limit: v.optional(boundedIntParam(1, BREAKDOWN_MAX_ROWS)), +}); + // Constrained natural-language query intent: the LLM only emits a value matching this schema, which // the executor maps onto existing aggregate helpers. Never used to build SQL from model text. export const QueryIntentSchema = v.object({ @@ -288,6 +336,7 @@ export type CollectInput = v.InferOutput; export type ServerEventInput = v.InferOutput; export type StatsQueryInput = v.InferOutput; export type DimensionSeriesQueryInput = v.InferOutput; +export type BreakdownQueryInput = v.InferOutput; export type CreateSiteInput = v.InferOutput; export type IssueKeyInput = v.InferOutput; export type GoalInput = v.InferOutput; diff --git a/packages/shared/src/stats.ts b/packages/shared/src/stats.ts index f8850f5..8597235 100644 --- a/packages/shared/src/stats.ts +++ b/packages/shared/src/stats.ts @@ -1,6 +1,6 @@ // Shared stats-API types: query parameters and response shapes for GET /api/stats. -import type { QueryIntent } from './schemas.js'; +import type { BreakdownDimension, QueryIntent } from './schemas.js'; /** Time-bucket granularity for time-series responses. */ export type Interval = 'hour' | 'day'; @@ -214,6 +214,38 @@ export interface StatsResponse { meta?: Freshness; } +/** One group of a `GET /api/stats/breakdown` response. `key` is the dimension's value, with an + * absent one reported as the empty string — the columnar store has no NULL, so both stores fold a + * missing dimension to `''` rather than each inventing its own label for it. */ +export interface BreakdownRow { + key: string; + /** All events in the group, pageviews included. */ + events: number; + pageviews: number; + /** Distinct visitor hashes within the group. NOT additive across groups, and bounded below by the + * k-anonymity floor every group had to clear to appear at all. */ + visitors: number; +} + +/** + * A single-dimension breakdown over the range. + * + * Which store answered is part of the response, not an implementation detail: the columnar store + * SAMPLES at high volume, so `events` and `pageviews` are sampling-corrected estimates and + * `visitors` — a distinct count, which no sampling weight can correct — is a LOWER bound whenever + * `sampled` is true. A caller that needs exact figures asks for a range D1 can serve, or reads the + * fixed top-N lists on `/api/stats`, which are always exact. + */ +export interface BreakdownResponse { + dimension: BreakdownDimension; + /** Which store produced these rows. */ + source: 'analytics_engine' | 'd1'; + /** True when the columnar store returned sampled rows, making every figure an estimate. Always + * false for `d1`, which scans every row. */ + sampled: boolean; + rows: BreakdownRow[]; +} + // ── Visualization contracts (distribution, per-dimension series, path hierarchy, journeys, clock) ── // // These five exist because the cube and the flat top-N lists cannot express the shapes a box plot, a