返回 CodeWhale
ingest.test.ts
根目录 / telemetry-ingest / test / ingest.test.ts
1 import { describe, expect, it } from "vitest";
2
3 import worker from "../src/index";
4 import { INGEST_PATH } from "../src/route";
5 import { BLOB_COLUMNS, DOUBLE_COLUMNS } from "../src/datapoint";
6 import { MAX_BODY_BYTES } from "../src/schema";
7 import { goldenBatch, harness, post, postJson } from "./support";
8
9 describe("method and route", () => {
10 it("answers 405 to GET, with no body", async () => {
11 const { env, written } = harness();
12 const response = await worker.fetch(
13 new Request(`https://telemetry.invalid${INGEST_PATH}`),
14 env,
15 );
16 expect(response.status).toBe(405);
17 expect(response.headers.get("allow")).toBe("POST");
18 expect(await response.text()).toBe("");
19 expect(written).toHaveLength(0);
20 });
21
22 it.each(["HEAD", "PUT", "PATCH", "DELETE", "OPTIONS"])(
23 "answers 405 to %s",
24 async (method) => {
25 const { env } = harness();
26 const response = await worker.fetch(
27 new Request(`https://telemetry.invalid${INGEST_PATH}`, { method }),
28 env,
29 );
30 expect(response.status).toBe(405);
31 },
32 );
33
34 it("has no readable surface: GET on any other path is also 405", async () => {
35 const { env } = harness();
36 for (const path of ["/", "/v1", "/v1/telemetry/query", "/health"]) {
37 const response = await worker.fetch(
38 new Request(`https://telemetry.invalid${path}`),
39 env,
40 );
41 expect(response.status).toBe(405);
42 expect(await response.text()).toBe("");
43 }
44 });
45
46 it("answers 404 to a POST at another path", async () => {
47 const { env } = harness();
48 const response = await worker.fetch(
49 post(JSON.stringify(goldenBatch()), { path: "/v1/other" }),
50 env,
51 );
52 expect(response.status).toBe(404);
53 });
54
55 it("answers 415 without a JSON content type", async () => {
56 const { env } = harness();
57 const response = await worker.fetch(
58 post(JSON.stringify(goldenBatch()), { contentType: "text/plain" }),
59 env,
60 );
61 expect(response.status).toBe(415);
62 });
63 });
64
65 describe("a valid batch", () => {
66 it("accepts the client's own golden v1 batch with 204 and no body", async () => {
67 const { env, written } = harness();
68 const response = await worker.fetch(postJson(goldenBatch()), env);
69
70 expect(response.status).toBe(204);
71 expect(await response.text()).toBe("");
72 expect(response.headers.get("content-type")).toBeNull();
73 expect(written).toHaveLength(4);
74 });
75
76 it("writes one data point per event, indexed by install_id only", async () => {
77 const { env, written } = harness();
78 await worker.fetch(postJson(goldenBatch()), env);
79
80 for (const point of written) {
81 expect(point.indexes).toEqual(["3f2a9c1e-0000-4000-8000-000000000001"]);
82 expect(point.blobs).toHaveLength(BLOB_COLUMNS.length);
83 expect(point.doubles).toHaveLength(DOUBLE_COLUMNS.length);
84 }
85 expect(written.map((point) => point.blobs[0])).toEqual([
86 "install_or_upgrade",
87 "session_start",
88 "session_end",
89 "panic",
90 ]);
91 });
92
93 it("puts the counters, errors and turn_wall in the documented columns", async () => {
94 const { env, written } = harness();
95 await worker.fetch(postJson(goldenBatch()), env);
96
97 const sessionEnd = written[2];
98 const column = (name: string) =>
99 sessionEnd.doubles[DOUBLE_COLUMNS.indexOf(name as never)];
100 expect(column("turns")).toBe(14);
101 expect(column("tool_calls")).toBe(61);
102 expect(column("subagent_spawn")).toBe(2);
103 expect(column("provider_http_5xx")).toBe(1);
104 expect(column("lt_5s")).toBe(9);
105 expect(column("gte_120s")).toBe(0);
106
107 const blob = (name: string) =>
108 sessionEnd.blobs[BLOB_COLUMNS.indexOf(name as never)];
109 expect(blob("providers")).toBe("custom,deepseek");
110 expect(blob("exit_class")).toBe("panic");
111 expect(blob("cold_start_bucket")).toBe("250_1000");
112
113 expect(written[3].blobs[BLOB_COLUMNS.indexOf("panic_site")]).toBe(
114 "crates/tui/src/tui/ui.rs:8801:17",
115 );
116 });
117
118 it("stores nothing that is not in the batch", async () => {
119 const { env, written } = harness();
120 await worker.fetch(postJson(goldenBatch()), env);
121
122 const stored = written.flatMap((point) => [
123 ...point.indexes,
124 ...point.blobs,
125 ]);
126 const batchText = JSON.stringify(goldenBatch());
127 for (const value of stored) {
128 if (value === "" || value === "true" || value === "false") continue;
129 // Every stored string is a substring of the batch the client sent, or a
130 // comma-join of values from it. Nothing is derived from the request.
131 for (const part of value.split(",")) {
132 expect(batchText).toContain(part);
133 }
134 }
135 });
136 });
137
138 describe("the closed field set", () => {
139 it("rejects an unknown key on the envelope", async () => {
140 const { env, written } = harness();
141 const batch = goldenBatch();
142 batch.cwd = "/Users/someone/work/secret-repo";
143 const response = await worker.fetch(postJson(batch), env);
144
145 expect(response.status).toBe(400);
146 expect(await response.text()).toBe("");
147 expect(written).toHaveLength(0);
148 });
149
150 it("rejects an unknown key nested inside an event", async () => {
151 const { env, written } = harness();
152 const batch = goldenBatch();
153 const events = batch.events as Record<string, unknown>[];
154 events[2].prompt = "please refactor crates/tui/src/main.rs";
155 const response = await worker.fetch(postJson(batch), env);
156
157 expect(response.status).toBe(400);
158 expect(written).toHaveLength(0);
159 });
160
161 it("rejects an unknown key nested inside counters", async () => {
162 const { env } = harness();
163 const batch = goldenBatch();
164 const events = batch.events as Record<string, Record<string, unknown>>[];
165 events[2].counters.git_branch_switches = 1;
166 expect((await worker.fetch(postJson(batch), env)).status).toBe(400);
167 });
168
169 it("rejects an unknown event discriminant", async () => {
170 const { env } = harness();
171 const batch = goldenBatch();
172 (batch.events as unknown[]).push({ event: "keystroke", site: "<dep>" });
173 expect((await worker.fetch(postJson(batch), env)).status).toBe(400);
174 });
175
176 it("rejects a batch with any envelope key removed", async () => {
177 const original = goldenBatch();
178 for (const key of Object.keys(original)) {
179 const { env } = harness();
180 const batch = goldenBatch();
181 delete batch[key];
182 const response = await worker.fetch(postJson(batch), env);
183 expect(response.status, `deleting ${key} must reject`).toBe(400);
184 }
185 });
186
187 it("rejects a batch with any event key removed", async () => {
188 const events = goldenBatch().events as Record<string, unknown>[];
189 for (let index = 0; index < events.length; index += 1) {
190 for (const key of Object.keys(events[index])) {
191 const { env } = harness();
192 const batch = goldenBatch();
193 delete (batch.events as Record<string, unknown>[])[index][key];
194 const response = await worker.fetch(postJson(batch), env);
195 expect(response.status, `deleting events[${index}].${key}`).toBe(400);
196 }
197 }
198 });
199 });
200
201 describe("value rules from the published schema", () => {
202 const reject = async (mutate: (batch: Record<string, unknown>) => void) => {
203 const { env, written } = harness();
204 const batch = goldenBatch();
205 mutate(batch);
206 const response = await worker.fetch(postJson(batch), env);
207 expect(response.status).toBe(400);
208 expect(written).toHaveLength(0);
209 };
210
211 it("rejects a wrong schema_version", () =>
212 reject((batch) => {
213 batch.schema_version = 2;
214 }));
215
216 it("rejects a non-UTC or sub-second sent_at", () =>
217 reject((batch) => {
218 batch.sent_at = "2026-08-03T18:04:11.234Z";
219 }));
220
221 it("rejects an install_id that is not a v4 uuid", () =>
222 reject((batch) => {
223 batch.install_id = "hostname-derived-id";
224 }));
225
226 it("rejects an app_version that is not a release version", () =>
227 reject((batch) => {
228 batch.app_version = "0.9.4 (/Users/someone/src/codewhale)";
229 }));
230
231 it("rejects a git_sha that is not 12 hex chars", () =>
232 reject((batch) => {
233 batch.git_sha = "refs/heads/feature-acme-migration";
234 }));
235
236 it("accepts a null git_sha — every locally built binary sends one", async () => {
237 const { env } = harness();
238 const batch = goldenBatch();
239 batch.git_sha = null;
240 expect((await worker.fetch(postJson(batch), env)).status).toBe(204);
241 });
242
243 it("accepts a null cold_start_bucket — non-TUI surfaces send one", async () => {
244 const { env } = harness();
245 const batch = goldenBatch();
246 (batch.events as Record<string, unknown>[])[2].cold_start_bucket = null;
247 expect((await worker.fetch(postJson(batch), env)).status).toBe(204);
248 });
249
250 it("rejects a surface outside the closed enum", () =>
251 reject((batch) => {
252 batch.surface = "desktop";
253 }));
254
255 it("rejects a provider list that is not sorted and deduplicated", () =>
256 reject((batch) => {
257 (batch.events as Record<string, unknown>[])[2].providers = [
258 "deepseek",
259 "custom",
260 ];
261 }));
262
263 it("rejects a provider entry shaped like a config table key", () =>
264 reject((batch) => {
265 (batch.events as Record<string, unknown>[])[2].providers = [
266 "acme_internal_gateway",
267 ];
268 }));
269
270 it("rejects a panic site outside the crates/ allowlist", () =>
271 reject((batch) => {
272 (batch.events as Record<string, unknown>[])[3].site =
273 "/Users/builder/.cargo/registry/src/index.crates.io/ratatui-0.29.0/src/lib.rs:1:1";
274 }));
275
276 it("accepts the reduced <dep> panic site", async () => {
277 const { env } = harness();
278 const batch = goldenBatch();
279 (batch.events as Record<string, unknown>[])[3].site = "<dep>";
280 expect((await worker.fetch(postJson(batch), env)).status).toBe(204);
281 });
282
283 it("rejects a counter that is not a non-negative integer", () =>
284 reject((batch) => {
285 const events = batch.events as Record<string, Record<string, unknown>>[];
286 events[2].counters.turns = -1;
287 }));
288
289 it("rejects more events than the client can put in one batch", () =>
290 reject((batch) => {
291 batch.events = Array.from({ length: 201 }, () => ({
292 event: "session_start",
293 source: "interactive",
294 }));
295 }));
296
297 it("rejects a body that is not JSON", async () => {
298 const { env } = harness();
299 const response = await worker.fetch(post("not json at all"), env);
300 expect(response.status).toBe(400);
301 });
302
303 it("rejects a JSON array at the top level", async () => {
304 const { env } = harness();
305 expect((await worker.fetch(postJson([goldenBatch()]), env)).status).toBe(
306 400,
307 );
308 });
309 });
310
311 describe("size caps", () => {
312 it("rejects an oversized body declared by content-length", async () => {
313 const { env, written } = harness();
314 const batch = goldenBatch();
315 batch.events = [
316 {
317 event: "panic",
318 site: `crates/tui/src/${"a".repeat(MAX_BODY_BYTES)}.rs:1:1`,
319 },
320 ];
321 const body = JSON.stringify(batch);
322 expect(body.length).toBeGreaterThan(MAX_BODY_BYTES);
323
324 const response = await worker.fetch(post(body), env);
325 expect(response.status).toBe(413);
326 expect(await response.text()).toBe("");
327 expect(written).toHaveLength(0);
328 });
329
330 it("rejects an oversized body that declares no content-length", async () => {
331 const { env, written } = harness();
332 const chunk = new TextEncoder().encode("x".repeat(8 * 1024));
333 let remaining = MAX_BODY_BYTES + 8 * 1024;
334 const stream = new ReadableStream<Uint8Array>({
335 pull(controller) {
336 if (remaining <= 0) {
337 controller.close();
338 return;
339 }
340 remaining -= chunk.byteLength;
341 controller.enqueue(chunk);
342 },
343 });
344
345 const request = new Request(`https://telemetry.invalid${INGEST_PATH}`, {
346 method: "POST",
347 headers: { "content-type": "application/json" },
348 body: stream,
349 // Node requires this for a streaming request body; Workers does not.
350 duplex: "half",
351 } as RequestInit);
352
353 expect(request.headers.get("content-length")).toBeNull();
354 const response = await worker.fetch(request, env);
355 expect(response.status).toBe(413);
356 expect(written).toHaveLength(0);
357 });
358
359 it("accepts a batch at the documented ceiling of 200 events", async () => {
360 const { env, written } = harness();
361 const batch = goldenBatch();
362 batch.events = Array.from({ length: 200 }, () => ({
363 event: "session_start",
364 source: "interactive",
365 }));
366 const response = await worker.fetch(postJson(batch), env);
367 expect(response.status).toBe(204);
368 expect(written).toHaveLength(200);
369 });
370 });
371
372 describe("rate limiting", () => {
373 it("keys the limiter on install_id and nothing else", async () => {
374 const { env, limited } = harness({ rateLimit: true });
375 const response = await worker.fetch(postJson(goldenBatch()), env);
376 expect(response.status).toBe(204);
377 expect(limited).toEqual(["3f2a9c1e-0000-4000-8000-000000000001"]);
378 });
379
380 it("answers 429 with no body when the limiter refuses", async () => {
381 const { env, written } = harness({ rateLimit: false });
382 const response = await worker.fetch(postJson(goldenBatch()), env);
383 expect(response.status).toBe(429);
384 expect(await response.text()).toBe("");
385 expect(written).toHaveLength(0);
386 });
387 });
388
389 describe("failing closed and quiet", () => {
390 it("answers 500 with no body when a binding throws", async () => {
391 const env = {
392 TELEMETRY: {
393 writeDataPoint() {
394 throw new Error("dataset unavailable");
395 },
396 },
397 };
398 const response = await worker.fetch(postJson(goldenBatch()), env);
399 expect(response.status).toBe(500);
400 expect(await response.text()).toBe("");
401 });
402
403 it("never returns a body on any path", async () => {
404 const { env } = harness();
405 const requests = [
406 new Request(`https://telemetry.invalid${INGEST_PATH}`),
407 post("{", {}),
408 post(JSON.stringify(goldenBatch()), { path: "/nope" }),
409 postJson(goldenBatch()),
410 ];
411 for (const request of requests) {
412 const response = await worker.fetch(request, env);
413 expect(await response.text()).toBe("");
414 }
415 });
416 });
417
417 lines TYPESCRIPT