| 1 | /** |
| 2 | * Batch -> Workers Analytics Engine data points. |
| 3 | * |
| 4 | * One data point per event. A batch is capped at 200 events |
| 5 | * (`BATCH_MAX_EVENTS`), and Analytics Engine allows 250 data points per Worker |
| 6 | * invocation, so a conforming batch never needs a second pass and never has to |
| 7 | * drop an event to fit. |
| 8 | * |
| 9 | * The column layout is fixed and positional because Analytics Engine columns |
| 10 | * are `blob1..blob20` / `double1..double20` — the names live only in the SQL |
| 11 | * you write. Renumbering a column silently rewrites every historical query, so |
| 12 | * the order below is append-only: to add a field, take the next free slot. |
| 13 | * |
| 14 | * The layout is chosen so the two questions the owner actually has are one |
| 15 | * query each. Both queries are written out in `README.md`: |
| 16 | * |
| 17 | * (a) installs and sessions -> index1 (install_id) + blob1 (event) |
| 18 | * (b) error classes and panic sites -> double11..16 + blob16, one GROUP BY |
| 19 | * |
| 20 | * Every value below comes out of the validated batch body and nothing else. |
| 21 | * This module cannot see anything about the connection — that is enforced by |
| 22 | * `test/no-ip.test.ts`, which fails if the type it takes is ever widened. See |
| 23 | * the red-line comment at the top of `index.ts`. |
| 24 | */ |
| 25 | |
| 26 | import type { Batch, Event } from "./schema"; |
| 27 | import { COUNTER_FIELDS, ERROR_FIELDS, TURN_WALL_FIELDS } from "./schema"; |
| 28 | |
| 29 | /** The subset of `AnalyticsEngineDataset` this Worker uses. */ |
| 30 | export interface DataPointSink { |
| 31 | writeDataPoint(point: { |
| 32 | indexes?: string[]; |
| 33 | blobs?: string[]; |
| 34 | doubles?: number[]; |
| 35 | }): void; |
| 36 | } |
| 37 | |
| 38 | /** One row, in Analytics Engine's positional form. */ |
| 39 | export interface DataPoint { |
| 40 | indexes: string[]; |
| 41 | blobs: string[]; |
| 42 | doubles: number[]; |
| 43 | } |
| 44 | |
| 45 | /** |
| 46 | * Blob column names, in `blob1..blobN` order. Exported so the README's SQL and |
| 47 | * the tests read from one list rather than two. |
| 48 | */ |
| 49 | export const BLOB_COLUMNS = [ |
| 50 | "event", // blob1 |
| 51 | "surface", // blob2 |
| 52 | "os", // blob3 |
| 53 | "arch", // blob4 |
| 54 | "libc", // blob5 |
| 55 | "app_version", // blob6 |
| 56 | "git_sha", // blob7 '' when null (a local build) |
| 57 | "tty", // blob8 'true' | 'false' |
| 58 | "install_kind", // blob9 |
| 59 | "previous_version", // blob10 |
| 60 | "session_source", // blob11 |
| 61 | "duration_bucket", // blob12 |
| 62 | "exit_class", // blob13 |
| 63 | "cold_start_bucket", // blob14 |
| 64 | "providers", // blob15 comma-joined, already sorted and deduplicated |
| 65 | "panic_site", // blob16 |
| 66 | "sent_at", // blob17 the batch timestamp; events carry none |
| 67 | "aggregate_counters", // blob18 closed product/operations JSON count object |
| 68 | "schema_version", // blob19 |
| 69 | "privacy_version", // blob20: v3 notice, v2 consent, empty on v1; blob19 disambiguates |
| 70 | ] as const; |
| 71 | |
| 72 | /** |
| 73 | * Double column names, in `double1..double20` order: the ten counters, the six |
| 74 | * error classes, then the four turn-wall buckets. Exactly 20 — Analytics |
| 75 | * Engine's ceiling — which is why `tty` is a blob. |
| 76 | */ |
| 77 | export const DOUBLE_COLUMNS = [ |
| 78 | ...COUNTER_FIELDS, |
| 79 | ...ERROR_FIELDS, |
| 80 | ...TURN_WALL_FIELDS, |
| 81 | ] as const; |
| 82 | |
| 83 | const EMPTY_DOUBLES: number[] = DOUBLE_COLUMNS.map(() => 0); |
| 84 | |
| 85 | /** Build the rows for one validated batch. */ |
| 86 | export function toDataPoints(batch: Batch): DataPoint[] { |
| 87 | return batch.events.map((event) => toDataPoint(batch, event)); |
| 88 | } |
| 89 | |
| 90 | function toDataPoint(batch: Batch, event: Event): DataPoint { |
| 91 | const blobs = new Array<string>(BLOB_COLUMNS.length).fill(""); |
| 92 | blobs[0] = event.event; |
| 93 | blobs[1] = batch.surface; |
| 94 | blobs[2] = batch.os; |
| 95 | blobs[3] = batch.arch; |
| 96 | blobs[4] = batch.libc; |
| 97 | blobs[5] = batch.app_version; |
| 98 | blobs[6] = batch.git_sha ?? ""; |
| 99 | blobs[7] = batch.tty ? "true" : "false"; |
| 100 | blobs[16] = batch.sent_at; |
| 101 | blobs[18] = String(batch.schema_version); |
| 102 | const privacyVersion = batch.schema_version === 3 ? batch.notice_version : batch.consent_version; |
| 103 | blobs[19] = privacyVersion === undefined ? "" : String(privacyVersion); |
| 104 | |
| 105 | let doubles = EMPTY_DOUBLES; |
| 106 | |
| 107 | switch (event.event) { |
| 108 | case "install_or_upgrade": |
| 109 | blobs[8] = event.kind; |
| 110 | blobs[9] = event.previous_version ?? ""; |
| 111 | break; |
| 112 | case "session_start": |
| 113 | blobs[10] = event.source; |
| 114 | break; |
| 115 | case "session_end": |
| 116 | blobs[11] = event.duration_bucket; |
| 117 | blobs[12] = event.exit_class; |
| 118 | blobs[13] = event.cold_start_bucket ?? ""; |
| 119 | blobs[14] = event.providers.join(","); |
| 120 | doubles = [ |
| 121 | ...COUNTER_FIELDS.map((field) => event.counters[field]), |
| 122 | ...ERROR_FIELDS.map((field) => event.errors[field]), |
| 123 | ...TURN_WALL_FIELDS.map((field) => event.turn_wall[field]), |
| 124 | ]; |
| 125 | break; |
| 126 | case "operations_summary": { |
| 127 | const { event: _event, ...counts } = event; |
| 128 | blobs[17] = JSON.stringify(counts); |
| 129 | break; |
| 130 | } |
| 131 | case "product_usage": |
| 132 | blobs[17] = JSON.stringify(event.counters); |
| 133 | break; |
| 134 | case "panic": |
| 135 | blobs[15] = event.site; |
| 136 | break; |
| 137 | } |
| 138 | |
| 139 | return { |
| 140 | // The one index. `install_id` is a random v4 UUID that the client rotates |
| 141 | // every 90 days, and it is the only identifier in the schema — which is |
| 142 | // also why `docs/TELEMETRY.md` says no count derived from it is a user |
| 143 | // count. It is the index because both questions group by it or count it. |
| 144 | indexes: [batch.install_id], |
| 145 | blobs, |
| 146 | doubles, |
| 147 | }; |
| 148 | } |
| 149 | |
| 150 | /** Write one batch. `writeDataPoint` is non-blocking and is never awaited. */ |
| 151 | export function writeBatch(sink: DataPointSink, batch: Batch): number { |
| 152 | const points = toDataPoints(batch); |
| 153 | for (const point of points) { |
| 154 | sink.writeDataPoint(point); |
| 155 | } |
| 156 | return points.length; |
| 157 | } |
| 158 |