* Batch -> Workers Analytics Engine data points.
*
* One data point per event. A batch is capped at 200 events
* (`BATCH_MAX_EVENTS`), and Analytics Engine allows 250 data points per Worker
* invocation, so a conforming batch never needs a second pass and never has to
* drop an event to fit.
*
* The column layout is fixed and positional because Analytics Engine columns
* are `blob1..blob20` / `double1..double20` — the names live only in the SQL
* you write. Renumbering a column silently rewrites every historical query, so
* the order below is append-only: to add a field, take the next free slot.
*
* The layout is chosen so the two questions the owner actually has are one
* query each. Both queries are written out in `README.md`:
*
* (a) installs and sessions -> index1 (install_id) + blob1 (event)
* (b) error classes and panic sites -> double11..16 + blob16, one GROUP BY
*
* Every value below comes out of the validated batch body and nothing else.
* This module cannot see anything about the connection — that is enforced by
* `test/no-ip.test.ts`, which fails if the type it takes is ever widened. See
* the red-line comment at the top of `index.ts`.
*/
import type { Batch, Event } from "./schema";
import { COUNTER_FIELDS, ERROR_FIELDS, TURN_WALL_FIELDS } from "./schema";
export interface DataPointSink {
writeDataPoint(point: {
indexes?: string[];
blobs?: string[];
doubles?: number[];
}): void;
}
export interface DataPoint {
indexes: string[];
blobs: string[];
doubles: number[];
}
* Blob column names, in `blob1..blobN` order. Exported so the README's SQL and
* the tests read from one list rather than two.
*/
export const BLOB_COLUMNS = [
"event",
"surface",
"os",
"arch",
"libc",
"app_version",
"git_sha",
"tty",
"install_kind",
"previous_version",
"session_source",
"duration_bucket",
"exit_class",
"cold_start_bucket",
"providers",
"panic_site",
"sent_at",
"aggregate_counters",
"schema_version",
"privacy_version",
] as const;
* Double column names, in `double1..double20` order: the ten counters, the six
* error classes, then the four turn-wall buckets. Exactly 20 — Analytics
* Engine's ceiling — which is why `tty` is a blob.
*/
export const DOUBLE_COLUMNS = [
...COUNTER_FIELDS,
...ERROR_FIELDS,
...TURN_WALL_FIELDS,
] as const;
const EMPTY_DOUBLES: number[] = DOUBLE_COLUMNS.map(() => 0);
export function toDataPoints(batch: Batch): DataPoint[] {
return batch.events.map((event) => toDataPoint(batch, event));
}
function toDataPoint(batch: Batch, event: Event): DataPoint {
const blobs = new Array<string>(BLOB_COLUMNS.length).fill("");
blobs[0] = event.event;
blobs[1] = batch.surface;
blobs[2] = batch.os;
blobs[3] = batch.arch;
blobs[4] = batch.libc;
blobs[5] = batch.app_version;
blobs[6] = batch.git_sha ?? "";
blobs[7] = batch.tty ? "true" : "false";
blobs[16] = batch.sent_at;
blobs[18] = String(batch.schema_version);
const privacyVersion = batch.schema_version === 3 ? batch.notice_version : batch.consent_version;
blobs[19] = privacyVersion === undefined ? "" : String(privacyVersion);
let doubles = EMPTY_DOUBLES;
switch (event.event) {
case "install_or_upgrade":
blobs[8] = event.kind;
blobs[9] = event.previous_version ?? "";
break;
case "session_start":
blobs[10] = event.source;
break;
case "session_end":
blobs[11] = event.duration_bucket;
blobs[12] = event.exit_class;
blobs[13] = event.cold_start_bucket ?? "";
blobs[14] = event.providers.join(",");
doubles = [
...COUNTER_FIELDS.map((field) => event.counters[field]),
...ERROR_FIELDS.map((field) => event.errors[field]),
...TURN_WALL_FIELDS.map((field) => event.turn_wall[field]),
];
break;
case "operations_summary": {
const { event: _event, ...counts } = event;
blobs[17] = JSON.stringify(counts);
break;
}
case "product_usage":
blobs[17] = JSON.stringify(event.counters);
break;
case "panic":
blobs[15] = event.site;
break;
}
return {
indexes: [batch.install_id],
blobs,
doubles,
};
}
export function writeBatch(sink: DataPointSink, batch: Batch): number {
const points = toDataPoints(batch);
for (const point of points) {
sink.writeDataPoint(point);
}
return points.length;
}