| 1 | /** |
| 2 | * Codewhale first-party telemetry ingest. |
| 3 | * |
| 4 | * ============================ THE RED LINE ============================ |
| 5 | * |
| 6 | * THIS WORKER NEVER READS, LOGS, STORES, OR FORWARDS THE CLIENT IP. |
| 7 | * |
| 8 | * `docs/TELEMETRY.md` publishes "Batches are IP-stripped at ingest. No IP is |
| 9 | * stored, logged, or joined to `install_id`." This file is the whole of what |
| 10 | * makes that sentence true. There is no other component. If you add an IP read |
| 11 | * here, the published document becomes a false statement about a shipped |
| 12 | * product, and every user who opted in did so on the strength of it. |
| 13 | * |
| 14 | * Concretely, and permanently: |
| 15 | * |
| 16 | * - Do not read the connecting-address header, the proxy-chain header, or any |
| 17 | * other header that carries a network address. The names are deliberately |
| 18 | * spelled nowhere in this directory: `test/no-ip.test.ts` greps the source |
| 19 | * for them and fails the build, so an edit that adds one cannot land quietly |
| 20 | * and cannot be justified as "just for debugging". |
| 21 | * - Do not read the `cf` property of the request. Country, colo, city, region, |
| 22 | * ASN, and coordinates all live there; none of them are in the schema, and |
| 23 | * the same test greps for that access too. |
| 24 | * - Read exactly two headers, ever: `content-type` and `content-length`. The |
| 25 | * same test extracts every header name this source asks for and fails if the |
| 26 | * set grows. |
| 27 | * - Do not log the request. Structured logs in Workers are queryable, and a |
| 28 | * log line is storage. Nothing in this file logs a payload or a header. |
| 29 | * - Analytics Engine rows are built in `datapoint.ts` from the *validated |
| 30 | * batch body only*. The request object is not in scope there. |
| 31 | * |
| 32 | * Debugging without an IP is a solved problem: the schema carries `os`, `arch`, |
| 33 | * `libc`, `surface`, `app_version`, and `git_sha`, which is what a crash triage |
| 34 | * actually needs. If you find yourself wanting the IP, you want a different |
| 35 | * feature, and it needs the owner's sign-off and a doc change first. |
| 36 | * |
| 37 | * ====================================================================== |
| 38 | * |
| 39 | * Shape of the service: write-only. One POST route, no GET that returns data, |
| 40 | * no response body on any path, ever. The client |
| 41 | * (`crates/telemetry/src/client.rs`) reads only the status class and drops the |
| 42 | * batch on anything that is not 2xx — no retry, no backoff, no re-queue — so a |
| 43 | * rejection here is invisible to the user by construction, and a 5xx can never |
| 44 | * become a client-visible error. That is what lets this endpoint fail closed: |
| 45 | * when in doubt, refuse the batch. |
| 46 | */ |
| 47 | |
| 48 | import { writeBatch, type DataPointSink } from "./datapoint"; |
| 49 | import { deliverPostHog, type PostHogConfig } from "./posthog"; |
| 50 | import { INGEST_PATH } from "./route"; |
| 51 | import { MAX_BODY_BYTES, validateBatch } from "./schema"; |
| 52 | |
| 53 | /** |
| 54 | * Rate-limit binding shape (`ratelimits` in `wrangler.jsonc`). |
| 55 | * |
| 56 | * Note that this module exports exactly one value — the default handler. The |
| 57 | * Workers runtime maps every *named* export of the entrypoint to an entrypoint |
| 58 | * of its own and refuses to start when one is not callable. Interfaces are |
| 59 | * erased at build time, so these cost nothing; a constant would not. |
| 60 | */ |
| 61 | export interface RateLimiter { |
| 62 | limit(options: { key: string }): Promise<{ success: boolean }>; |
| 63 | } |
| 64 | |
| 65 | export interface Env extends PostHogConfig { |
| 66 | /** `analytics_engine_datasets` binding. */ |
| 67 | TELEMETRY: DataPointSink; |
| 68 | /** |
| 69 | * Optional per-install rate limiter. |
| 70 | * |
| 71 | * Keyed on `install_id` — the identifier the batch already carries — and |
| 72 | * never on a network address. That is a weaker limiter than an IP-keyed one |
| 73 | * (a `install_id.json` can be rewritten between POSTs) and it is the right |
| 74 | * trade: Cloudflare's edge already absorbs volumetric abuse, and the failure |
| 75 | * mode of an IP-keyed limiter is that this Worker starts handling IPs. |
| 76 | */ |
| 77 | RATE_LIMITER?: RateLimiter; |
| 78 | } |
| 79 | |
| 80 | /** Every response is a bare status. No body, no echo of the payload, ever. */ |
| 81 | function status(code: number, headers?: HeadersInit): Response { |
| 82 | return new Response(null, { status: code, headers }); |
| 83 | } |
| 84 | |
| 85 | /** |
| 86 | * Read at most `limit` bytes, aborting the stream the moment it goes over. |
| 87 | * |
| 88 | * `content-length` is checked first as a cheap reject, but it is client-supplied |
| 89 | * and may be absent or wrong, so the real bound is enforced while reading. |
| 90 | * Returns `null` when the body is missing or too large. |
| 91 | */ |
| 92 | async function readBounded( |
| 93 | body: ReadableStream<Uint8Array> | null, |
| 94 | limit: number, |
| 95 | ): Promise<Uint8Array | null> { |
| 96 | if (body === null) return null; |
| 97 | const reader = body.getReader(); |
| 98 | const chunks: Uint8Array[] = []; |
| 99 | let total = 0; |
| 100 | try { |
| 101 | for (;;) { |
| 102 | const { done, value } = await reader.read(); |
| 103 | if (done) break; |
| 104 | if (value === undefined) continue; |
| 105 | total += value.byteLength; |
| 106 | if (total > limit) { |
| 107 | await reader.cancel(); |
| 108 | return null; |
| 109 | } |
| 110 | chunks.push(value); |
| 111 | } |
| 112 | } finally { |
| 113 | reader.releaseLock(); |
| 114 | } |
| 115 | const joined = new Uint8Array(total); |
| 116 | let offset = 0; |
| 117 | for (const chunk of chunks) { |
| 118 | joined.set(chunk, offset); |
| 119 | offset += chunk.byteLength; |
| 120 | } |
| 121 | return joined; |
| 122 | } |
| 123 | |
| 124 | async function ingest(request: Request, env: Env): Promise<Response> { |
| 125 | // Method before path, so a probe of any path with any verb other than POST |
| 126 | // gets the same answer and learns nothing about what exists here. |
| 127 | if (request.method !== "POST") { |
| 128 | return status(405, { allow: "POST" }); |
| 129 | } |
| 130 | if (new URL(request.url).pathname !== INGEST_PATH) { |
| 131 | return status(404); |
| 132 | } |
| 133 | |
| 134 | // Header read #1 of 2. `client.rs` sends exactly `application/json`. |
| 135 | const contentType = request.headers.get("content-type") ?? ""; |
| 136 | if (!contentType.toLowerCase().startsWith("application/json")) { |
| 137 | return status(415); |
| 138 | } |
| 139 | |
| 140 | // Header read #2 of 2, and the last. See the red line above. |
| 141 | const declared = request.headers.get("content-length"); |
| 142 | if (declared !== null) { |
| 143 | const length = Number(declared); |
| 144 | if (!Number.isFinite(length) || length > MAX_BODY_BYTES) { |
| 145 | return status(413); |
| 146 | } |
| 147 | } |
| 148 | |
| 149 | const raw = await readBounded(request.body, MAX_BODY_BYTES); |
| 150 | if (raw === null) return status(413); |
| 151 | |
| 152 | let text: string; |
| 153 | try { |
| 154 | text = new TextDecoder("utf-8", { fatal: true }).decode(raw); |
| 155 | } catch { |
| 156 | return status(400); |
| 157 | } |
| 158 | |
| 159 | let parsed: unknown; |
| 160 | try { |
| 161 | parsed = JSON.parse(text); |
| 162 | } catch { |
| 163 | return status(400); |
| 164 | } |
| 165 | |
| 166 | // The closed-field-set check. An unexpected key anywhere rejects the whole |
| 167 | // batch: a future client bug that starts attaching a path or a prompt must be |
| 168 | // refused by the server rather than quietly stored. The reason string stays |
| 169 | // here — the response carries no body, because a parse error echoed back is a |
| 170 | // way to learn what this endpoint keeps. |
| 171 | const result = validateBatch(parsed); |
| 172 | if (!result.ok) return status(400); |
| 173 | |
| 174 | // Rate limiting is keyed on the install id the batch already carries. It runs |
| 175 | // after validation because that is the only way to have a non-network key. |
| 176 | if (env.RATE_LIMITER !== undefined) { |
| 177 | const { success } = await env.RATE_LIMITER.limit({ |
| 178 | key: result.batch.install_id, |
| 179 | }); |
| 180 | if (!success) return status(429); |
| 181 | } |
| 182 | |
| 183 | // `writeDataPoint` is non-blocking and returns void; it is never awaited. |
| 184 | writeBatch(env.TELEMETRY, result.batch); |
| 185 | |
| 186 | // The optional processor sees only the same validated batch, never the |
| 187 | // incoming request. Its bounded, best-effort failure cannot reject AE data. |
| 188 | await deliverPostHog(result.batch, env); |
| 189 | |
| 190 | return status(204); |
| 191 | } |
| 192 | |
| 193 | export default { |
| 194 | async fetch(request: Request, env: Env): Promise<Response> { |
| 195 | try { |
| 196 | return await ingest(request, env); |
| 197 | } catch { |
| 198 | // Fail closed and quiet. Nothing is written, nothing is logged, and the |
| 199 | // response has no body. The client treats any non-2xx as "dropped" and |
| 200 | // surfaces nothing to the user, so a 5xx here costs one batch and never |
| 201 | // becomes a user-visible error. |
| 202 | return status(500); |
| 203 | } |
| 204 | }, |
| 205 | }; |
| 206 |