1import * as NodeHttpServer from "@effect/platform-node/NodeHttpServer";
2import * as NodeCrypto from "@effect/platform-node/NodeCrypto";
3import * as NodeServices from "@effect/platform-node/NodeServices";
4import { assert, it } from "@effect/vitest";
5import * as Clock from "effect/Clock";
6import * as ConfigProvider from "effect/ConfigProvider";
7import * as Effect from "effect/Effect";
8import * as Exit from "effect/Exit";
9import * as Fiber from "effect/Fiber";
10import * as Layer from "effect/Layer";
11import * as Schema from "effect/Schema";
12import * as Scope from "effect/Scope";
13import * as TestClock from "effect/testing/TestClock";
14import * as HttpClient from "effect/http/HttpClient";
15import * as HttpClientError from "effect/http/HttpClientError";
16import * as HttpClientRequest from "effect/http/HttpClientRequest";
17import * as HttpClientResponse from "effect/http/HttpClientResponse";
18import * as HttpServer from "effect/http/HttpServer";
19import * as HttpServerRequest from "effect/http/HttpServerRequest";
20import * as HttpServerResponse from "effect/http/HttpServerResponse";
21import * as HostProcess from "@t3tools/shared/HostProcess";
22
23import type { ManagedPolicyDocument } from "@t3tools/contracts";
24
25import * as ServerConfig from "../config.ts";
26import * as ManagedPolicy from "../policy/ManagedPolicy.ts";
27import { getTelemetryIdentity } from "./Identify.ts";
28import * as AnalyticsService from "./AnalyticsService.ts";
29
30interface RecordedBatchRequest {
31 readonly path: string;
32 readonly body: {
33 readonly batch?: ReadonlyArray<{
34 readonly event?: string;
35 readonly properties?: {
36 readonly index?: number;
37 readonly clientType?: string;
38 readonly serverOs?: string;
39 readonly serverArch?: string;
40 readonly serverAppVersion?: string;
41 readonly serverMode?: string;
42 readonly t3CodeVersion?: string;
43 };
44 }>;
45 } | null;
46}
47
48interface RecordedBatchBody {
49 readonly batch: ReadonlyArray<{
50 readonly event?: string;
51 readonly properties?: {
52 readonly index?: number;
53 readonly clientType?: string;
54 readonly serverOs?: string;
55 readonly serverArch?: string;
56 readonly serverAppVersion?: string;
57 readonly serverMode?: string;
58 readonly t3CodeVersion?: string;
59 };
60 }>;
61}
62
63const SentBatch = Schema.fromJsonString(
64 Schema.Struct({
65 batch: Schema.Array(
66 Schema.Struct({
67 uuid: Schema.String,
68 event: Schema.String,
69 distinct_id: Schema.String,
70 properties: Schema.Struct({
71 telemetryChannel: Schema.String,
72 installId: Schema.optionalKey(Schema.String),
73 ownerSource: Schema.optionalKey(Schema.String),
74 observedBy: Schema.optionalKey(Schema.String),
75 }),
76 }),
77 ),
78 }),
79);
80const decodeSentBatch = Schema.decodeEffect(SentBatch);
81type SentEvent = (typeof SentBatch.Type)["batch"][number];
82type Respond = (
83 request: HttpClientRequest.HttpClientRequest,
84) => Effect.Effect<HttpClientResponse.HttpClientResponse, HttpClientError.HttpClientError>;
85
86const respondWithStatus =
87 (status: number): Respond =>
88 (request) =>
89 Effect.succeed(HttpClientResponse.fromWeb(request, new Response(null, { status })));
90
91/**
92 * Answers 503 for `minutes` after the first send, then 204. `answers` holds each
93 * answer with its time in ms after the first send (the test clock runs on across tests).
94 */
95const failingFor = (minutes: number) => {
96 const answers: Array<{ readonly at: number; readonly status: number }> = [];
97 let firstSend: number | undefined;
98 const respond: Respond = (request) =>
99 Effect.flatMap(Clock.currentTimeMillis, (now) => {
100 firstSend ??= now;
101 const at = now - firstSend;
102 const status = at < minutes * 60_000 ? 503 : 204;
103 answers.push({ at, status });
104 return respondWithStatus(status)(request);
105 });
106 return { respond, answers };
107};
108
109const recordTwentyEvents = Effect.gen(function* () {
110 const analytics = yield* AnalyticsService.AnalyticsService;
111 for (let index = 0; index < 20; index += 1) {
112 yield* analytics.record("test.retry", { index });
113 }
114});
115
116const analyticsLayerWith = (client: HttpClient.HttpClient, policy?: ManagedPolicyDocument) =>
117 AnalyticsService.layer.pipe(
118 Layer.provideMerge(
119 ServerConfig.ServerConfig.layerTest(process.cwd(), { prefix: "t3-telemetry-retry-" }),
120 ),
121 Layer.provide(
122 ConfigProvider.layer(
123 ConfigProvider.fromUnknown({
124 T3CODE_TELEMETRY_ENABLED: true,
125 T3CODE_POSTHOG_KEY: "phc_test_key",
126 T3CODE_POSTHOG_HOST: "http://localhost",
127 }),
128 ),
129 ),
130 Layer.provide(
131 Layer.mergeAll(
132 Layer.succeed(HostProcess.Platform, "win32"),
133 Layer.succeed(HostProcess.Architecture, "x64"),
134 Layer.succeed(HttpClient.HttpClient, client),
135 ManagedPolicy.layerTest(policy),
136 ),
137 ),
138 );
139
140/**
141 * Runs `record` (20 events by default), then lets `minutes` of background
142 * flushes run against a client that answers every send with `respond`.
143 * Returns the sent batches.
144 */
145const sendFor = (
146 minutes: number,
147 respond: Respond,
148 record: Effect.Effect<
149 void,
150 never,
151 AnalyticsService.AnalyticsService | Effect.Services<typeof getTelemetryIdentity>
152 > = recordTwentyEvents,
153 policy?: ManagedPolicyDocument,
154) =>
155 Effect.gen(function* () {
156 const batches: Array<ReadonlyArray<SentEvent>> = [];
157 const client = HttpClient.make((request) =>
158 Effect.gen(function* () {
159 if (request.body._tag === "Uint8Array") {
160 const body = yield* decodeSentBatch(new TextDecoder().decode(request.body.body)).pipe(
161 Effect.orDie,
162 );
163 batches.push(body.batch);
164 }
165 return yield* respond(request);
166 }),
167 );
168 const runtimeLayer = analyticsLayerWith(client, policy);
169
170 yield* Effect.gen(function* () {
171 yield* record;
172 for (let second = 0; second < minutes * 60; second += 1) {
173 yield* TestClock.adjust("1 second");
174 }
175 }).pipe(Effect.provide(runtimeLayer));
176 return batches;
177 });
178
179it.layer(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer))("AnalyticsService test", (it) => {
180 it.effect("waits for desktop shell consent and honors a late telemetry opt-out", () =>
181 Effect.gen(function* () {
182 const env: NodeJS.ProcessEnv = { T3CODE_DESKTOP_SHELL_DISCOVERY_PENDING: "1" };
183 const batches = yield* sendFor(
184 0,
185 respondWithStatus(204),
186 Effect.gen(function* () {
187 const analytics = yield* AnalyticsService.AnalyticsService;
188 yield* analytics.record("test.before-discovery");
189 yield* analytics.flush;
190 env.T3CODE_TELEMETRY_ENABLED = "false";
191 delete env.T3CODE_DESKTOP_SHELL_DISCOVERY_PENDING;
192 yield* analytics.record("test.opted-out");
193 yield* analytics.flush;
194 env.T3CODE_TELEMETRY_ENABLED = "true";
195 yield* analytics.record("test.allowed");
196 yield* analytics.flush;
197 }),
198 ).pipe(Effect.provideService(HostProcess.Environment, env));
199 assert.deepEqual(
200 batches.flat().map((event) => event.event),
201 ["test.allowed"],
202 );
203 }),
204 );
205
206 it.effect("client owner reports cannot use up the server's owner cap", () =>
207 Effect.gen(function* () {
208 const batches = yield* sendFor(
209 10,
210 respondWithStatus(204),
211 Effect.gen(function* () {
212 const analytics = yield* AnalyticsService.AnalyticsService;
213 yield* analytics.recordOwners(
214 Array.from({ length: 1_001 }, (_, index) => ({
215 source: "device" as const,
216 ownerId: index.toString(16).padStart(64, "0"),
217 })),
218 "client",
219 );
220 // The event buffer also holds 1,000, so send the client events first.
221 yield* analytics.flush;
222 yield* analytics.recordOwners([{ source: "connect", ownerId: "a".repeat(64) }], "server");
223 }),
224 );
225 const owners = batches.flat().map((event) => event.properties.observedBy);
226
227 assert.equal(owners.filter((observedBy) => observedBy === "client").length, 1_000);
228 assert.equal(owners.filter((observedBy) => observedBy === "server").length, 1);
229 }),
230 );
231
232 it.effect("keeps a failed batch through a 20-minute outage and sends it once after", () =>
233 Effect.gen(function* () {
234 const ingest = failingFor(20);
235 const batches = yield* sendFor(40, ingest.respond);
236 const statuses = ingest.answers.map(({ status }) => status);
237 assert.equal(statuses.at(-1), 204);
238 assert.equal(statuses.filter((status) => status === 204).length, 1);
239 const sentUuids = new Set(batches.map((batch) => batch.map(({ uuid }) => uuid).join()));
240 assert.equal(sentUuids.size, 1);
241 assert.equal(batches[0]?.length, 20);
242 }),
243 );
244
245 it.effect("drops a batch that keeps failing for 30 minutes", () =>
246 Effect.gen(function* () {
247 const ingest = failingFor(Infinity);
248 yield* sendFor(60, ingest.respond);
249 const lastSend = ingest.answers.at(-1)?.at ?? 0;
250 assert.isAtLeast(lastSend, 30 * 60_000);
251 assert.isBelow(lastSend, 36 * 60_000);
252 }),
253 );
254
255 it.effect(
256 "sends nothing when a managed policy turns telemetry off, despite the environment",
257 () =>
258 Effect.gen(function* () {
259 const policy = { schemaVersion: 1, organization: "Example Corp" } as const;
260 const allowed = yield* sendFor(1, respondWithStatus(204), recordTwentyEvents, {
261 ...policy,
262 limits: { telemetry: true },
263 });
264 assert.equal(allowed.length, 1);
265 const denied = yield* sendFor(1, respondWithStatus(204), recordTwentyEvents, {
266 ...policy,
267 limits: { telemetry: false },
268 });
269 assert.deepEqual(denied, []);
270 }),
271 );
272
273 it.effect("drops a batch the endpoint rejects without sending it again", () =>
274 Effect.gen(function* () {
275 const batches = yield* sendFor(10, respondWithStatus(400));
276 assert.equal(batches.length, 1);
277 }),
278 );
279
280 it.effect("shutdown stops waiting for a telemetry endpoint that never answers", () =>
281 Effect.gen(function* () {
282 const scope = yield* Scope.make();
283 const context = yield* Layer.buildWithScope(
284 analyticsLayerWith(HttpClient.make(() => Effect.never)),
285 scope,
286 );
287 yield* recordTwentyEvents.pipe(Effect.provide(context));
288
289 const shutdown = yield* Scope.close(scope, Exit.void).pipe(Effect.forkChild);
290 yield* TestClock.adjust("1 second");
291 yield* Fiber.join(shutdown);
292 }),
293 );
294
295 it.effect("flush drains all buffered events across multiple batches", () =>
296 Effect.gen(function* () {
297 const capturedRequests: Array<RecordedBatchRequest> = [];
298 const layerServerConfig = ServerConfig.ServerConfig.layerTest(process.cwd(), {
299 prefix: "t3-telemetry-base-",
300 });
301
302 const layerTelemetry = AnalyticsService.layer.pipe(Layer.provideMerge(layerServerConfig));
303 const layerConfig = ConfigProvider.layer(
304 ConfigProvider.fromUnknown({
305 T3CODE_TELEMETRY_ENABLED: true,
306 T3CODE_POSTHOG_KEY: "phc_test_key",
307 T3CODE_POSTHOG_HOST: "http://localhost",
308 T3CODE_TELEMETRY_FLUSH_BATCH_SIZE: 20,
309 }),
310 );
311 const layerBatchServer = HttpServer.serve(
312 Effect.gen(function* () {
313 const request = yield* HttpServerRequest.HttpServerRequest;
314 if (request.method !== "POST") {
315 return HttpServerResponse.empty({ status: 404 });
316 }
317
318 const payload = yield* request.json.pipe(
319 Effect.map((body) => body as RecordedBatchRequest["body"]),
320 Effect.orElseSucceed(() => null),
321 );
322
323 capturedRequests.push({ path: request.url, body: payload });
324
325 return HttpServerResponse.jsonUnsafe({});
326 }),
327 );
328 const layerRuntime = layerTelemetry.pipe(
329 Layer.provide(layerConfig),
330 Layer.provide(
331 Layer.mergeAll(
332 Layer.succeed(HostProcess.Platform, "linux"),
333 Layer.succeed(HostProcess.Architecture, "arm64"),
334 ManagedPolicy.layerTest(),
335 ),
336 ),
337 Layer.provideMerge(NodeHttpServer.layerTest),
338 );
339
340 yield* Effect.gen(function* () {
341 yield* Layer.launch(layerBatchServer).pipe(Effect.forkScoped);
342 const { distinctId } = yield* getTelemetryIdentity;
343 assert.equal(distinctId !== null, true);
344 const analytics = yield* AnalyticsService.AnalyticsService;
345
346 for (let index = 0; index < 45; index += 1) {
347 yield* analytics.record("test.flush.drain", { index });
348 }
349
350 yield* analytics.flush;
351 }).pipe(Effect.provide(layerRuntime));
352
353 const batchRequests = capturedRequests.filter(
354 (request): request is RecordedBatchRequest & { readonly body: RecordedBatchBody } =>
355 Array.isArray(request.body?.batch),
356 );
357 assert.equal(batchRequests.length, 3);
358 assert.equal(
359 batchRequests.every(
360 (request) => request.path.endsWith("/batch/") || request.path.endsWith("/batch"),
361 ),
362 true,
363 );
364 const deliveredIndexes = batchRequests.flatMap((request) =>
365 request.body.batch
366 .filter((event) => event.event === "test.flush.drain")
367 .map((event) => event.properties?.index)
368 .filter((index): index is number => typeof index === "number"),
369 );
370
371 const sorted = deliveredIndexes.toSorted((a, b) => a - b);
372 assert.equal(sorted.length, 45);
373 assert.deepEqual(
374 sorted,
375 Array.from({ length: 45 }, (_, index) => index),
376 );
377 assert.equal(
378 batchRequests.every((request) =>
379 request.body.batch.every((event) => event.properties?.clientType === "cli-web-client"),
380 ),
381 true,
382 );
383 assert.equal(
384 batchRequests.every((request) =>
385 request.body.batch.every(
386 (event) =>
387 event.properties?.serverOs === "Linux" &&
388 event.properties.serverArch === "arm64" &&
389 event.properties.serverAppVersion === event.properties.t3CodeVersion &&
390 event.properties.serverMode === "web",
391 ),
392 ),
393 true,
394 );
395 }),
396 );
397
398 it.effect("does not send batch requests when telemetry is disabled", () =>
399 Effect.gen(function* () {
400 const capturedPaths: Array<string> = [];
401 const layerServerConfig = ServerConfig.ServerConfig.layerTest(process.cwd(), {
402 prefix: "t3-telemetry-disabled-",
403 });
404 const layerTelemetry = AnalyticsService.layer.pipe(Layer.provideMerge(layerServerConfig));
405 const layerConfig = ConfigProvider.layer(
406 ConfigProvider.fromUnknown({
407 T3CODE_TELEMETRY_ENABLED: false,
408 T3CODE_POSTHOG_KEY: "phc_test_key",
409 T3CODE_POSTHOG_HOST: "http://localhost",
410 }),
411 );
412 const layerBatchServer = HttpServer.serve(
413 Effect.gen(function* () {
414 const request = yield* HttpServerRequest.HttpServerRequest;
415 capturedPaths.push(request.url);
416 return HttpServerResponse.jsonUnsafe({});
417 }),
418 );
419 const layerRuntime = layerTelemetry.pipe(
420 Layer.provide(layerConfig),
421 Layer.provide(
422 Layer.mergeAll(
423 Layer.succeed(HostProcess.Platform, "linux"),
424 Layer.succeed(HostProcess.Architecture, "arm64"),
425 ManagedPolicy.layerTest(),
426 ),
427 ),
428 Layer.provideMerge(NodeHttpServer.layerTest),
429 );
430
431 yield* Effect.gen(function* () {
432 yield* Layer.launch(layerBatchServer).pipe(Effect.forkScoped);
433 const analytics = yield* AnalyticsService.AnalyticsService;
434 yield* analytics.record("test.disabled", { index: 1 });
435 yield* analytics.flush;
436 }).pipe(Effect.provide(layerRuntime));
437
438 assert.deepEqual(capturedPaths, []);
439 }),
440 );
441});