/** * Anonymous PostHog telemetry service. * * Persists an installation-scoped anonymous identifier, buffers events in * memory, and flushes batches over Effect's HTTP client. * * @module AnalyticsService */import { HostProcessArchitecture, HostProcessPlatform } from "@t3tools/shared/hostProcess";import { getUrlDiagnostics } from "@t3tools/shared/urlDiagnostics";import * as Config from "effect/Config";import * as Context from "effect/Context";import * as DateTime from "effect/DateTime";import * as Effect from "effect/Effect";import * as Layer from "effect/Layer";import * as Option from "effect/Option";import * as Ref from "effect/Ref";import * as Schema from "effect/Schema";import * as HttpClient from "effect/unstable/http/HttpClient";import * as HttpClientRequest from "effect/unstable/http/HttpClientRequest";import * as HttpClientResponse from "effect/unstable/http/HttpClientResponse"; import packageJson from "../../package.json" with { type: "json" };import * as ServerConfig from "../config.ts";import { getTelemetryIdentifier } from "./Identify.ts"; interface BufferedAnalyticsEvent { readonly event: string; readonly properties?: Readonly<Record<string, unknown>>; readonly capturedAt: string;} const TelemetryEnvConfig = Config.all({ posthogKey: Config.string("T3CODE_POSTHOG_KEY").pipe( Config.withDefault("phc_XOWci4oZP4VvLiEyrFqkFjP4CZn55mjYYBMREK5Wd6m"), ), posthogHost: Config.string("T3CODE_POSTHOG_HOST").pipe( Config.withDefault("https://us.i.posthog.com"), ), enabled: Config.boolean("T3CODE_TELEMETRY_ENABLED").pipe(Config.withDefault(true)), flushBatchSize: Config.number("T3CODE_TELEMETRY_FLUSH_BATCH_SIZE").pipe(Config.withDefault(20)), maxBufferedEvents: Config.number("T3CODE_TELEMETRY_MAX_BUFFERED_EVENTS").pipe( Config.withDefault(1_000), ), wslDistroName: Config.string("WSL_DISTRO_NAME").pipe(Config.option),}); export class AnalyticsBatchDeliveryError extends Schema.TaggedErrorClass<AnalyticsBatchDeliveryError>()( "AnalyticsBatchDeliveryError", { endpointInputLength: Schema.Number, endpointProtocol: Schema.optionalKey(Schema.String), endpointHostname: Schema.optionalKey(Schema.String), eventCount: Schema.Int.check(Schema.isGreaterThan(0)), cause: Schema.Defect(), },) { static fromEndpoint(input: { readonly endpoint: string; readonly eventCount: number; readonly cause: unknown; }): AnalyticsBatchDeliveryError { const diagnostics = getUrlDiagnostics(input.endpoint); return new AnalyticsBatchDeliveryError({ endpointInputLength: diagnostics.inputLength, ...(diagnostics.protocol === undefined ? {} : { endpointProtocol: diagnostics.protocol }), ...(diagnostics.hostname === undefined ? {} : { endpointHostname: diagnostics.hostname }), eventCount: input.eventCount, cause: input.cause, }); } override get message(): string { const eventLabel = this.eventCount === 1 ? "event" : "events"; const destination = this.endpointHostname ? ` at ${this.endpointHostname}` : ""; return `Failed to deliver ${this.eventCount} analytics ${eventLabel} to PostHog${destination} (endpoint input length ${this.endpointInputLength}).`; }} export class AnalyticsService extends Context.Service< AnalyticsService, { /** Record an anonymous event for best-effort buffered delivery. */ readonly record: ( event: string, properties?: Readonly<Record<string, unknown>>, ) => Effect.Effect<void>; /** Flush all currently queued telemetry events. */ readonly flush: Effect.Effect<void>; }>()("t3/telemetry/AnalyticsService") { /** No-op layer for callers that intentionally disable telemetry. */ static readonly layerTest = Layer.succeed( AnalyticsService, AnalyticsService.of({ record: () => Effect.void, flush: Effect.void, }), );} export const make = Effect.gen(function* () { const telemetryConfig = yield* TelemetryEnvConfig; const httpClient = yield* HttpClient.HttpClient; const serverConfig = yield* ServerConfig.ServerConfig; const identifier = yield* getTelemetryIdentifier; const bufferRef = yield* Ref.make<ReadonlyArray<BufferedAnalyticsEvent>>([]); const clientType = serverConfig.mode === "desktop" ? "desktop-app" : "cli-web-client"; const hostPlatform = yield* HostProcessPlatform; const hostArchitecture = yield* HostProcessArchitecture; const batchEndpoint = `${telemetryConfig.posthogHost}/batch/`; const enqueueBufferedEvent = (event: string, properties?: Readonly<Record<string, unknown>>) => Effect.flatMap(DateTime.now, (now) => Ref.modify(bufferRef, (current) => { const appended = [ ...current, { event, ...(properties ? { properties } : {}), capturedAt: DateTime.formatIso(now), } satisfies BufferedAnalyticsEvent, ]; const next = appended.length > telemetryConfig.maxBufferedEvents ? appended.slice(appended.length - telemetryConfig.maxBufferedEvents) : appended; return [ { size: next.length, dropped: next.length !== appended.length, } as const, next, ] as const; }), ); const sendBatch = Effect.fn("AnalyticsService.sendBatch")(function* ( events: ReadonlyArray<BufferedAnalyticsEvent>, ) { if (!telemetryConfig.enabled || !identifier) return; const payload = { api_key: telemetryConfig.posthogKey, batch: events.map((event) => ({ event: event.event, distinct_id: identifier, properties: { ...event.properties, $process_person_profile: false, platform: hostPlatform, wsl: Option.getOrUndefined(telemetryConfig.wslDistroName), arch: hostArchitecture, t3CodeVersion: packageJson.version, clientType, }, timestamp: event.capturedAt, })), }; yield* HttpClientRequest.post(batchEndpoint).pipe( HttpClientRequest.bodyJson(payload), Effect.flatMap(httpClient.execute), Effect.flatMap(HttpClientResponse.filterStatusOk), Effect.mapError((cause) => AnalyticsBatchDeliveryError.fromEndpoint({ endpoint: batchEndpoint, eventCount: events.length, cause, }), ), ); }); const flush: AnalyticsService["Service"]["flush"] = Effect.gen(function* () { while (true) { const batch = yield* Ref.modify(bufferRef, (current) => { if (current.length === 0) { return [[] as ReadonlyArray<BufferedAnalyticsEvent>, current] as const; } const nextBatch = current.slice(0, telemetryConfig.flushBatchSize); const remaining = current.slice(nextBatch.length); return [nextBatch, remaining] as const; }); if (batch.length === 0) { return; } yield* sendBatch(batch).pipe( Effect.catchTags({ AnalyticsBatchDeliveryError: (error) => Ref.update(bufferRef, (current) => [...batch, ...current]).pipe( Effect.flatMap(() => Effect.fail(error)), ), }), ); } }).pipe( Effect.catchTags({ AnalyticsBatchDeliveryError: (error) => Effect.logError(error.message).pipe( Effect.annotateLogs({ endpointInputLength: error.endpointInputLength, ...(error.endpointProtocol === undefined ? {} : { endpointProtocol: error.endpointProtocol }), ...(error.endpointHostname === undefined ? {} : { endpointHostname: error.endpointHostname }), eventCount: error.eventCount, cause: error, }), ), }), ); const record: AnalyticsService["Service"]["record"] = Effect.fn("AnalyticsService.record")( function* (event, properties) { if (!telemetryConfig.enabled || !identifier) return; const enqueueResult = yield* enqueueBufferedEvent(event, properties); if (enqueueResult.dropped) { yield* Effect.logDebug("analytics buffer full; dropping oldest event", { size: enqueueResult.size, event, }); } }, ); yield* Effect.forever(Effect.sleep(1000).pipe(Effect.flatMap(() => flush)), { disableYield: true, }).pipe(Effect.forkScoped); yield* Effect.addFinalizer(() => flush); return AnalyticsService.of({ record, flush });}); export const layer = Layer.effect(AnalyticsService, make); export const layerTest = AnalyticsService.layerTest;
/**
* Anonymous PostHog telemetry service.
*
* Persists an installation-scoped anonymous identifier, buffers events in
* memory, and flushes batches over Effect's HTTP client.
*
* @module AnalyticsService
*/
import { HostProcessArchitecture, HostProcessPlatform } from "@t3tools/shared/hostProcess";
import { getUrlDiagnostics } from "@t3tools/shared/urlDiagnostics";
import * as Config from "effect/Config";
import * as Context from "effect/Context";
import * as DateTime from "effect/DateTime";
import * as Effect from "effect/Effect";
import * as Layer from "effect/Layer";
import * as Option from "effect/Option";
import * as Ref from "effect/Ref";
import * as Schema from "effect/Schema";
import * as HttpClient from "effect/unstable/http/HttpClient";
import * as HttpClientRequest from "effect/unstable/http/HttpClientRequest";
import * as HttpClientResponse from "effect/unstable/http/HttpClientResponse";
import packageJson from "../../package.json" with { type: "json" };
import * as ServerConfig from "../config.ts";
import { getTelemetryIdentifier } from "./Identify.ts";
interface BufferedAnalyticsEvent {
readonly event: string;
readonly properties?: Readonly<Record<string, unknown>>;
readonly capturedAt: string;
}
const TelemetryEnvConfig = Config.all({
posthogKey: Config.string("T3CODE_POSTHOG_KEY").pipe(
Config.withDefault("phc_XOWci4oZP4VvLiEyrFqkFjP4CZn55mjYYBMREK5Wd6m"),
),
posthogHost: Config.string("T3CODE_POSTHOG_HOST").pipe(
Config.withDefault("https://us.i.posthog.com"),
),
enabled: Config.boolean("T3CODE_TELEMETRY_ENABLED").pipe(Config.withDefault(true)),
flushBatchSize: Config.number("T3CODE_TELEMETRY_FLUSH_BATCH_SIZE").pipe(Config.withDefault(20)),
maxBufferedEvents: Config.number("T3CODE_TELEMETRY_MAX_BUFFERED_EVENTS").pipe(
Config.withDefault(1_000),
),
wslDistroName: Config.string("WSL_DISTRO_NAME").pipe(Config.option),
});
export class AnalyticsBatchDeliveryError extends Schema.TaggedErrorClass<AnalyticsBatchDeliveryError>()(
"AnalyticsBatchDeliveryError",
{
endpointInputLength: Schema.Number,
endpointProtocol: Schema.optionalKey(Schema.String),
endpointHostname: Schema.optionalKey(Schema.String),
eventCount: Schema.Int.check(Schema.isGreaterThan(0)),
cause: Schema.Defect(),
},
) {
static fromEndpoint(input: {
readonly endpoint: string;
readonly eventCount: number;
readonly cause: unknown;
}): AnalyticsBatchDeliveryError {
const diagnostics = getUrlDiagnostics(input.endpoint);
return new AnalyticsBatchDeliveryError({
endpointInputLength: diagnostics.inputLength,
...(diagnostics.protocol === undefined ? {} : { endpointProtocol: diagnostics.protocol }),
...(diagnostics.hostname === undefined ? {} : { endpointHostname: diagnostics.hostname }),
eventCount: input.eventCount,
cause: input.cause,
});
}
override get message(): string {
const eventLabel = this.eventCount === 1 ? "event" : "events";
const destination = this.endpointHostname ? ` at ${this.endpointHostname}` : "";
return `Failed to deliver ${this.eventCount} analytics ${eventLabel} to PostHog${destination} (endpoint input length ${this.endpointInputLength}).`;
}
}
export class AnalyticsService extends Context.Service<
AnalyticsService,
{
/** Record an anonymous event for best-effort buffered delivery. */
readonly record: (
event: string,
properties?: Readonly<Record<string, unknown>>,
) => Effect.Effect<void>;
/** Flush all currently queued telemetry events. */
readonly flush: Effect.Effect<void>;
}
>()("t3/telemetry/AnalyticsService") {
/** No-op layer for callers that intentionally disable telemetry. */
static readonly layerTest = Layer.succeed(
AnalyticsService,
AnalyticsService.of({
record: () => Effect.void,
flush: Effect.void,
}),
);
}
export const make = Effect.gen(function* () {
const telemetryConfig = yield* TelemetryEnvConfig;
const httpClient = yield* HttpClient.HttpClient;
const serverConfig = yield* ServerConfig.ServerConfig;
const identifier = yield* getTelemetryIdentifier;
const bufferRef = yield* Ref.make<ReadonlyArray<BufferedAnalyticsEvent>>([]);
const clientType = serverConfig.mode === "desktop" ? "desktop-app" : "cli-web-client";
const hostPlatform = yield* HostProcessPlatform;
const hostArchitecture = yield* HostProcessArchitecture;
const batchEndpoint = `${telemetryConfig.posthogHost}/batch/`;
const enqueueBufferedEvent = (event: string, properties?: Readonly<Record<string, unknown>>) =>
Effect.flatMap(DateTime.now, (now) =>
Ref.modify(bufferRef, (current) => {
const appended = [
...current,
{
event,
...(properties ? { properties } : {}),
capturedAt: DateTime.formatIso(now),
} satisfies BufferedAnalyticsEvent,
];
const next =
appended.length > telemetryConfig.maxBufferedEvents
? appended.slice(appended.length - telemetryConfig.maxBufferedEvents)
: appended;
return [
{
size: next.length,
dropped: next.length !== appended.length,
} as const,
next,
] as const;
}),
);
const sendBatch = Effect.fn("AnalyticsService.sendBatch")(function* (
events: ReadonlyArray<BufferedAnalyticsEvent>,
) {
if (!telemetryConfig.enabled || !identifier) return;
const payload = {
api_key: telemetryConfig.posthogKey,
batch: events.map((event) => ({
event: event.event,
distinct_id: identifier,
properties: {
...event.properties,
$process_person_profile: false,
platform: hostPlatform,
wsl: Option.getOrUndefined(telemetryConfig.wslDistroName),
arch: hostArchitecture,
t3CodeVersion: packageJson.version,
clientType,
},
timestamp: event.capturedAt,
})),
};
yield* HttpClientRequest.post(batchEndpoint).pipe(
HttpClientRequest.bodyJson(payload),
Effect.flatMap(httpClient.execute),
Effect.flatMap(HttpClientResponse.filterStatusOk),
Effect.mapError((cause) =>
AnalyticsBatchDeliveryError.fromEndpoint({
endpoint: batchEndpoint,
eventCount: events.length,
cause,
}),
),
);
});
const flush: AnalyticsService["Service"]["flush"] = Effect.gen(function* () {
while (true) {
const batch = yield* Ref.modify(bufferRef, (current) => {
if (current.length === 0) {
return [[] as ReadonlyArray<BufferedAnalyticsEvent>, current] as const;
}
const nextBatch = current.slice(0, telemetryConfig.flushBatchSize);
const remaining = current.slice(nextBatch.length);
return [nextBatch, remaining] as const;
});
if (batch.length === 0) {
return;
}
yield* sendBatch(batch).pipe(
Effect.catchTags({
AnalyticsBatchDeliveryError: (error) =>
Ref.update(bufferRef, (current) => [...batch, ...current]).pipe(
Effect.flatMap(() => Effect.fail(error)),
),
}),
);
}
}).pipe(
Effect.catchTags({
AnalyticsBatchDeliveryError: (error) =>
Effect.logError(error.message).pipe(
Effect.annotateLogs({
endpointInputLength: error.endpointInputLength,
...(error.endpointProtocol === undefined
? {}
: { endpointProtocol: error.endpointProtocol }),
...(error.endpointHostname === undefined
? {}
: { endpointHostname: error.endpointHostname }),
eventCount: error.eventCount,
cause: error,
}),
),
}),
);
const record: AnalyticsService["Service"]["record"] = Effect.fn("AnalyticsService.record")(
function* (event, properties) {
if (!telemetryConfig.enabled || !identifier) return;
const enqueueResult = yield* enqueueBufferedEvent(event, properties);
if (enqueueResult.dropped) {
yield* Effect.logDebug("analytics buffer full; dropping oldest event", {
size: enqueueResult.size,
event,
});
}
},
);
yield* Effect.forever(Effect.sleep(1000).pipe(Effect.flatMap(() => flush)), {
disableYield: true,
}).pipe(Effect.forkScoped);
yield* Effect.addFinalizer(() => flush);
return AnalyticsService.of({ record, flush });
});
export const layer = Layer.effect(AnalyticsService, make);
export const layerTest = AnalyticsService.layerTest;