1// T3 ingest retains events and forwards to PostHog; UUIDs deduplicate retries there.
2import * as HostProcess from "@t3tools/shared/HostProcess";
3import type { ClientOs } from "@t3tools/contracts";
4import * as Clock from "effect/Clock";
5import * as Config from "effect/Config";
6import * as ConfigProvider from "effect/ConfigProvider";
7import * as Context from "effect/Context";
8import * as SecureRandom from "effect/crypto/SecureRandom";
9import * as DateTime from "effect/DateTime";
10import * as Effect from "effect/Effect";
11import * as Exit from "effect/Exit";
12import * as Layer from "effect/Layer";
13import * as Option from "effect/Option";
14import * as Random from "effect/Random";
15import * as Ref from "effect/Ref";
16import * as Result from "effect/Result";
17import * as Semaphore from "effect/Semaphore";
18import * as HttpClient from "effect/http/HttpClient";
19import * as HttpClientRequest from "effect/http/HttpClientRequest";
20import * as HttpClientResponse from "effect/http/HttpClientResponse";
21
22import packageJson from "../../package.json" with { type: "json" };
23import * as ServerConfig from "../config.ts";
24import * as ManagedPolicy from "../policy/ManagedPolicy.ts";
25import { getTelemetryIdentity, type TelemetryOwner } from "./Identify.ts";
26
27interface BufferedAnalyticsEvent {
28 readonly uuid: string;
29 readonly event: string;
30 readonly properties?: Readonly<Record<string, unknown>>;
31 readonly capturedAt: string;
32}
33
34interface DeliveryState {
35 readonly failedBatch: ReadonlyArray<BufferedAnalyticsEvent>;
36 readonly batchAttempts: number;
37 readonly firstFailedAt: number;
38 readonly failures: number;
39 readonly retryAt: number;
40}
41
42// Retry delays shorter than this still wait for the next background check.
43const FLUSH_INTERVAL_MS = 10_000;
44// A hung send would hold the flush lock for the next background flush.
45const SEND_TIMEOUT = "10 seconds";
46// Quit and restart wait for this flush; events it cannot send in time are dropped.
47const SHUTDOWN_FLUSH_TIMEOUT = "1 second";
48// A failed batch is tried again until it has failed for this long. Ingest answers 503 when it
49// cannot store a batch, and a broken Cloudflare location can take about 5 minutes to fail over.
50// The waiting batch holds newer events back; `maxBufferedEvents` bounds them meanwhile.
51const MAX_BATCH_RETRY_MS = 30 * 60_000;
52const RETRY_BASE_DELAY_MS = 2_000;
53// Clients can report any number of owner ids; past this many per observer, new ones
54// are not sent. Each observer has its own cap, so client reports never crowd out the
55// server's own.
56const MAX_RECORDED_OWNERS = 1_000;
57const RETRY_MAX_DELAY_MS = 300_000;
58// These statuses reject the batch itself; unlike 408/429, retrying cannot help.
59const REJECTED_BATCH_STATUSES = new Set([400, 401, 413]);
60
61/** `random` is in [0, 1). */
62function retryDelayMs(failures: number, random: number): number {
63 const ceiling = Math.min(RETRY_MAX_DELAY_MS, RETRY_BASE_DELAY_MS * 2 ** (failures - 1));
64 return Math.round(ceiling / 2 + (ceiling / 2) * random);
65}
66
67const TelemetryEnvConfig = Config.all({
68 posthogKey: Config.String("T3CODE_POSTHOG_KEY").pipe(
69 Config.withDefault("phc_XOWci4oZP4VvLiEyrFqkFjP4CZn55mjYYBMREK5Wd6m"),
70 ),
71 posthogHost: Config.String("T3CODE_POSTHOG_HOST").pipe(
72 Config.withDefault("https://ingest.t3.codes"),
73 ),
74 enabled: Config.Boolean("T3CODE_TELEMETRY_ENABLED").pipe(Config.withDefault(true)),
75 flushBatchSize: Config.Number("T3CODE_TELEMETRY_FLUSH_BATCH_SIZE").pipe(Config.withDefault(20)),
76 maxBufferedEvents: Config.Number("T3CODE_TELEMETRY_MAX_BUFFERED_EVENTS").pipe(
77 Config.withDefault(1_000),
78 ),
79 wslDistroName: Config.String("WSL_DISTRO_NAME").pipe(Config.option),
80});
81
82export class AnalyticsService extends Context.Service<
83 AnalyticsService,
84 {
85 readonly record: (
86 event: string,
87 properties?: Readonly<Record<string, unknown>>,
88 ) => Effect.Effect<void>;
89
90 readonly flush: Effect.Effect<void>;
91
92 /**
93 * Records `install.owner_observed` for each owner of this install that the
94 * server read itself or a connecting client reported. Each owner is sent once
95 * per server process, so reconnects and repeated reads are free.
96 */
97 readonly recordOwners: (
98 owners: ReadonlyArray<TelemetryOwner>,
99 observedBy: "server" | "client",
100 ) => Effect.Effect<void>;
101 }
102>()("t3/telemetry/AnalyticsService") {}
103
104function serverOsFromNodePlatform(platform: string): ClientOs {
105 switch (platform) {
106 case "darwin":
107 return "macOS";
108 case "win32":
109 return "Windows";
110 case "linux":
111 return "Linux";
112 case "android":
113 return "Android";
114 default:
115 return "other";
116 }
117}
118
119/** @public */
120/** @public */
121export const make = Effect.gen(function* () {
122 const telemetryConfig = yield* TelemetryEnvConfig;
123 const hostEnvironment = yield* HostProcess.Environment;
124 const awaitingShellConsent = hostEnvironment.T3CODE_DESKTOP_SHELL_DISCOVERY_PENDING === "1";
125 // A managed policy can switch analytics off over the environment, never on.
126 const policyAllows =
127 ManagedPolicy.deniedBy(yield* ManagedPolicy.ManagedPolicy, "telemetry") === undefined;
128 // A cached PATH does not establish consent from the current login shell.
129 const telemetryEnabled = Effect.suspend(() => {
130 if (!telemetryConfig.enabled || !policyAllows) return Effect.succeed(false);
131 if (!awaitingShellConsent) return Effect.succeed(true);
132 if (hostEnvironment.T3CODE_DESKTOP_SHELL_DISCOVERY_PENDING === "1")
133 return Effect.succeed(false);
134 const preference = hostEnvironment.T3CODE_TELEMETRY_ENABLED;
135 return Config.Boolean("T3CODE_TELEMETRY_ENABLED").pipe(
136 Config.withDefault(true),
137 Effect.provideService(
138 ConfigProvider.ConfigProvider,
139 ConfigProvider.fromEnv({
140 env: preference === undefined ? {} : { T3CODE_TELEMETRY_ENABLED: preference },
141 }),
142 ),
143 Effect.orElseSucceed(() => false),
144 );
145 });
146 const httpClient = yield* HttpClient.HttpClient;
147 const serverConfig = yield* ServerConfig.ServerConfig;
148 const { distinctId, installId } = yield* getTelemetryIdentity;
149 const crypto = yield* SecureRandom.SecureRandom;
150 const bufferRef = yield* Ref.make<ReadonlyArray<BufferedAnalyticsEvent>>([]);
151 const deliveryRef = yield* Ref.make<DeliveryState>({
152 failedBatch: [],
153 batchAttempts: 0,
154 firstFailedAt: 0,
155 failures: 0,
156 retryAt: 0,
157 });
158 const recordedOwnersRef = yield* Ref.make<
159 Readonly<Record<"server" | "client", ReadonlySet<string>>>
160 >({ server: new Set(), client: new Set() });
161 // The background flush and the shutdown flush must not send the same batch at once.
162 const flushLock = yield* Semaphore.make(1);
163 const clientType = serverConfig.mode === "desktop" ? "desktop-app" : "cli-web-client";
164 const hostPlatform = yield* HostProcess.Platform;
165 const hostArchitecture = yield* HostProcess.Architecture;
166
167 const enqueueBufferedEvent = (
168 uuid: string,
169 event: string,
170 properties?: Readonly<Record<string, unknown>>,
171 ) =>
172 Effect.flatMap(DateTime.now, (now) =>
173 Ref.modify(bufferRef, (current) => {
174 const appended = [
175 ...current,
176 {
177 uuid,
178 event,
179 ...(properties ? { properties } : {}),
180 capturedAt: DateTime.formatIso(now),
181 } satisfies BufferedAnalyticsEvent,
182 ];
183
184 const next =
185 appended.length > telemetryConfig.maxBufferedEvents
186 ? appended.slice(appended.length - telemetryConfig.maxBufferedEvents)
187 : appended;
188
189 return [
190 {
191 size: next.length,
192 dropped: next.length !== appended.length,
193 } as const,
194 next,
195 ] as const;
196 }),
197 );
198
199 const sendBatch = Effect.fn("AnalyticsService.sendBatch")(function* (
200 events: ReadonlyArray<BufferedAnalyticsEvent>,
201 ) {
202 if (!(yield* telemetryEnabled) || !distinctId) return;
203
204 const payload = {
205 api_key: telemetryConfig.posthogKey,
206 batch: events.map((event) => ({
207 uuid: event.uuid,
208 event: event.event,
209 distinct_id: distinctId,
210 properties: {
211 ...event.properties,
212 ...(installId === null ? {} : { installId }),
213 $process_person_profile: false,
214 platform: hostPlatform,
215 wsl: Option.getOrUndefined(telemetryConfig.wslDistroName),
216 arch: hostArchitecture,
217 t3CodeVersion: packageJson.version,
218 telemetryChannel: ServerConfig.telemetryChannel,
219 clientType,
220 serverOs: serverOsFromNodePlatform(hostPlatform),
221 serverArch: hostArchitecture,
222 serverWslDistro: Option.getOrUndefined(telemetryConfig.wslDistroName),
223 serverAppVersion: packageJson.version,
224 serverMode: serverConfig.mode,
225 },
226 timestamp: event.capturedAt,
227 })),
228 };
229
230 yield* HttpClientRequest.post(`${telemetryConfig.posthogHost}/batch/`).pipe(
231 HttpClientRequest.bodyJson(payload),
232 Effect.flatMap(httpClient.execute),
233 Effect.flatMap(HttpClientResponse.filterStatusOk),
234 Effect.timeout(SEND_TIMEOUT),
235 );
236 });
237
238 const takeBatch = Ref.modify(bufferRef, (current) => {
239 const nextBatch = current.slice(0, telemetryConfig.flushBatchSize);
240 return [nextBatch, current.slice(nextBatch.length)] as const;
241 });
242
243 // Back off even after dropping rejected batches so bad API keys cannot drain the queue in a tight loop.
244 const flush: AnalyticsService["Service"]["flush"] = Effect.gen(function* () {
245 while (true) {
246 const delivery = yield* Ref.get(deliveryRef);
247 const batch = delivery.failedBatch.length > 0 ? delivery.failedBatch : yield* takeBatch;
248 if (batch.length === 0) {
249 return;
250 }
251
252 const sent = yield* Effect.result(sendBatch(batch));
253 if (Result.isSuccess(sent)) {
254 yield* Ref.set(deliveryRef, {
255 failedBatch: [],
256 batchAttempts: 0,
257 firstFailedAt: 0,
258 failures: 0,
259 retryAt: 0,
260 });
261 continue;
262 }
263
264 const now = yield* Clock.currentTimeMillis;
265 const failures = delivery.failures + 1;
266 const batchAttempts = delivery.batchAttempts + 1;
267 const firstFailedAt = delivery.batchAttempts === 0 ? now : delivery.firstFailedAt;
268 const retryAt = now + retryDelayMs(failures, yield* Random.next);
269 const error = sent.failure;
270 const rejected =
271 error._tag === "HttpClientError" &&
272 error.reason._tag === "StatusCodeError" &&
273 REJECTED_BATCH_STATUSES.has(error.reason.response.status);
274 if (!rejected && now - firstFailedAt < MAX_BATCH_RETRY_MS) {
275 yield* Ref.set(deliveryRef, {
276 failedBatch: batch,
277 batchAttempts,
278 firstFailedAt,
279 failures,
280 retryAt,
281 });
282 yield* Effect.logDebug("Failed to send telemetry batch; will retry", {
283 attempt: batchAttempts,
284 cause: error,
285 });
286 return;
287 }
288 yield* Ref.set(deliveryRef, {
289 failedBatch: [],
290 batchAttempts: 0,
291 firstFailedAt: 0,
292 failures,
293 retryAt,
294 });
295 yield* Effect.logWarning(
296 rejected
297 ? "Dropped telemetry batch the endpoint rejected"
298 : "Dropped telemetry batch after repeated send failures",
299 { events: batch.length, attempts: batchAttempts, cause: error },
300 );
301 return;
302 }
303 }).pipe(flushLock.withPermit);
304
305 const flushWhenDue = Effect.gen(function* () {
306 const { retryAt } = yield* Ref.get(deliveryRef);
307 if ((yield* Clock.currentTimeMillis) >= retryAt) {
308 yield* flush;
309 }
310 });
311
312 const record: AnalyticsService["Service"]["record"] = Effect.fn("AnalyticsService.record")(
313 function* (event, properties) {
314 if (!(yield* telemetryEnabled) || !distinctId) return;
315
316 // Catch Node UUID defects too; best-effort telemetry drops events without UUIDs.
317 const uuid = yield* Effect.exit(crypto.uuidV7());
318 if (Exit.isFailure(uuid)) return;
319
320 const enqueueResult = yield* enqueueBufferedEvent(uuid.value, event, properties);
321 if (enqueueResult.dropped) {
322 yield* Effect.logDebug("analytics buffer full; dropping oldest event", {
323 size: enqueueResult.size,
324 event,
325 });
326 }
327 },
328 );
329
330 const recordOwners: AnalyticsService["Service"]["recordOwners"] = Effect.fn(
331 "AnalyticsService.recordOwners",
332 )((owners, observedBy) =>
333 Effect.forEach(
334 owners,
335 ({ source, ownerId }) =>
336 Ref.modify(recordedOwnersRef, (recorded) => {
337 const key = `${source}:${ownerId}`;
338 const seen = recorded[observedBy];
339 return seen.has(key) || seen.size >= MAX_RECORDED_OWNERS
340 ? [false, recorded]
341 : [true, { ...recorded, [observedBy]: new Set(seen).add(key) }];
342 }).pipe(
343 Effect.flatMap((isNew) =>
344 isNew
345 ? record("install.owner_observed", { ownerSource: source, ownerId, observedBy })
346 : Effect.void,
347 ),
348 ),
349 { discard: true },
350 ),
351 );
352
353 yield* Effect.forever(Effect.sleep(FLUSH_INTERVAL_MS).pipe(Effect.andThen(flushWhenDue)), {
354 disableYield: true,
355 }).pipe(Effect.forkScoped);
356
357 yield* Effect.addFinalizer(() =>
358 flush.pipe(Effect.timeout(SHUTDOWN_FLUSH_TIMEOUT), Effect.ignore),
359 );
360
361 return AnalyticsService.of({ record, flush, recordOwners });
362});
363
364export const layer = Layer.effect(AnalyticsService, make);