| author | |
| committer | |
| log | ec3778bbf4a5d8726cff22bddfcba82cd1c6ebde |
| tree | ef2fb0f61750f7065b8ace19e92a15003b248724 |
| parent | 8c5b8d3b720a82de1c77fe3aa32bde7a5ff36325 |
- introduce `AnalyticsService` with buffered PostHog batch flush
- add hashed telemetry identifier from Codex account id or anonymous local id
- record provider session/turn lifecycle events and flush on stop-all
- wire telemetry layer into server runtime and update tests with no-op analytics12 files changed, 337 insertions(+), 7 deletions(-)
apps/server/integration/OrchestrationEngineHarness.integration.ts+13| ... | ... | @@ -40,6 +40,7 @@ import { makeProviderServiceLive } from "../src/provider/Layers/ProviderService. |
| 40 | 40 | import { makeCodexAdapterLive } from "../src/provider/Layers/CodexAdapter.ts"; |
| 41 | 41 | import { CodexAdapter } from "../src/provider/Services/CodexAdapter.ts"; |
| 42 | 42 | import { ProviderService } from "../src/provider/Services/ProviderService.ts"; |
| 43 | import { AnalyticsService } from "../src/telemetry/Services/AnalyticsService.ts"; | |
| 43 | 44 | import { CheckpointReactorLive } from "../src/orchestration/Layers/CheckpointReactor.ts"; |
| 44 | 45 | import { OrchestrationEngineLive } from "../src/orchestration/Layers/OrchestrationEngine.ts"; |
| 45 | 46 | import { OrchestrationProjectionPipelineLive } from "../src/orchestration/Layers/ProjectionPipeline.ts"; |
| ... | ... | @@ -252,10 +253,22 @@ export const makeOrchestrationIntegrationHarness = ( |
| 252 | 253 | ? makeProviderServiceLive().pipe( |
| 253 | 254 | Layer.provide(providerSessionDirectoryLayer), |
| 254 | 255 | Layer.provide(realCodexRegistry), |
| 256 | Layer.provide( | |
| 257 | Layer.succeed(AnalyticsService, { | |
| 258 | record: () => Effect.void, | |
| 259 | flush: Effect.void, | |
| 260 | }), | |
| 261 | ), | |
| 255 | 262 | ) |
| 256 | 263 | : makeProviderServiceLive().pipe( |
| 257 | 264 | Layer.provide(providerSessionDirectoryLayer), |
| 258 | 265 | Layer.provide(fakeRegistry!), |
| 266 | Layer.provide( | |
| 267 | Layer.succeed(AnalyticsService, { | |
| 268 | record: () => Effect.void, | |
| 269 | flush: Effect.void, | |
| 270 | }), | |
| 271 | ), | |
| 259 | 272 | ); |
| 260 | 273 | |
| 261 | 274 | const runtimeServicesLayer = Layer.mergeAll( |
apps/server/integration/providerService.integration.test.ts+5| ... | ... | @@ -12,6 +12,7 @@ import { |
| 12 | 12 | ProviderService, |
| 13 | 13 | type ProviderServiceShape, |
| 14 | 14 | } from "../src/provider/Services/ProviderService.ts"; |
| 15 | import { AnalyticsService } from "../src/telemetry/Services/AnalyticsService.ts"; | |
| 15 | 16 | import { SqlitePersistenceMemory } from "../src/persistence/Layers/Sqlite.ts"; |
| 16 | 17 | import { ProviderSessionRuntimeRepositoryLive } from "../src/persistence/Layers/ProviderSessionRuntime.ts"; |
| 17 | 18 | |
| ... | ... | @@ -59,6 +60,10 @@ const makeIntegrationFixture = Effect.gen(function* () { |
| 59 | 60 | const shared = Layer.mergeAll( |
| 60 | 61 | directoryLayer, |
| 61 | 62 | Layer.succeed(ProviderAdapterRegistry, registry), |
| 63 | Layer.succeed(AnalyticsService, { | |
| 64 | record: () => Effect.void, | |
| 65 | flush: Effect.void, | |
| 66 | }), | |
| 62 | 67 | ).pipe(Layer.provide(SqlitePersistenceMemory)); |
| 63 | 68 | |
| 64 | 69 | const layer = makeProviderServiceLive().pipe(Layer.provide(shared)); |
apps/server/src/index.ts+4| ... | ... | @@ -9,13 +9,17 @@ import { Command } from "effect/unstable/cli"; |
| 9 | 9 | import { version } from "../package.json" with { type: "json" }; |
| 10 | 10 | import { ServerLive } from "./wsServer"; |
| 11 | 11 | import { NetService } from "@t3tools/shared/Net"; |
| 12 | import { AnalyticsServiceLayerLive } from "./telemetry/Layers/AnalyticsService"; | |
| 13 | import * as NodeHttpClient from "@effect/platform-node/NodeHttpClient"; | |
| 12 | 14 | |
| 13 | 15 | const RuntimeLayer = Layer.empty.pipe( |
| 14 | 16 | Layer.provideMerge(CliConfig.layer), |
| 17 | Layer.provideMerge(AnalyticsServiceLayerLive), | |
| 15 | 18 | Layer.provideMerge(ServerLive), |
| 16 | 19 | Layer.provideMerge(OpenLive), |
| 17 | 20 | Layer.provideMerge(NetService.layer), |
| 18 | 21 | Layer.provideMerge(NodeServices.layer), |
| 22 | Layer.provideMerge(NodeHttpClient.layerUndici), | |
| 19 | 23 | ); |
| 20 | 24 | |
| 21 | 25 | Command.run(t3Cli, { version }).pipe(Effect.provide(RuntimeLayer), NodeRuntime.runMain); |
apps/server/src/main.test.ts+5| ... | ... | @@ -11,6 +11,7 @@ import { NetService } from "@t3tools/shared/Net"; |
| 11 | 11 | import { CliConfig, t3Cli, type CliConfigShape } from "./main"; |
| 12 | 12 | import { ServerConfig, type ServerConfigShape } from "./config"; |
| 13 | 13 | import { Open, type OpenShape } from "./open"; |
| 14 | import { AnalyticsService } from "./telemetry/Services/AnalyticsService"; | |
| 14 | 15 | import { Server, type ServerShape } from "./wsServer"; |
| 15 | 16 | |
| 16 | 17 | const start = vi.fn(() => undefined); |
| ... | ... | @@ -47,6 +48,10 @@ const testLayer = Layer.mergeAll( |
| 47 | 48 | openBrowser: (_target: string) => Effect.void, |
| 48 | 49 | openInEditor: () => Effect.void, |
| 49 | 50 | } satisfies OpenShape), |
| 51 | Layer.succeed(AnalyticsService, { | |
| 52 | record: () => Effect.void, | |
| 53 | flush: Effect.void, | |
| 54 | }), | |
| 50 | 55 | NodeServices.layer, |
| 51 | 56 | ); |
| 52 | 57 |
apps/server/src/provider/Layers/ProviderService.test.ts+11| ... | ... | @@ -42,6 +42,7 @@ import { |
| 42 | 42 | makeSqlitePersistenceLive, |
| 43 | 43 | SqlitePersistenceMemory, |
| 44 | 44 | } from "../../persistence/Layers/Sqlite.ts"; |
| 45 | import { AnalyticsService } from "../../telemetry/Services/AnalyticsService.ts"; | |
| 45 | 46 | |
| 46 | 47 | const asRequestId = (value: string): ApprovalRequestId => ApprovalRequestId.makeUnsafe(value); |
| 47 | 48 | const asEventId = (value: string): EventId => EventId.makeUnsafe(value); |
| ... | ... | @@ -214,6 +215,11 @@ function makeFakeCodexAdapter(provider: ProviderKind = "codex") { |
| 214 | 215 | const sleep = (ms: number) => |
| 215 | 216 | Effect.promise(() => new Promise<void>((resolve) => setTimeout(resolve, ms))); |
| 216 | 217 | |
| 218 | const noopAnalyticsLayer = Layer.succeed(AnalyticsService, { | |
| 219 | record: () => Effect.void, | |
| 220 | flush: Effect.void, | |
| 221 | }); | |
| 222 | ||
| 217 | 223 | function makeProviderServiceLayer() { |
| 218 | 224 | const codex = makeFakeCodexAdapter(); |
| 219 | 225 | const registry: typeof ProviderAdapterRegistry.Service = { |
| ... | ... | @@ -235,8 +241,10 @@ function makeProviderServiceLayer() { |
| 235 | 241 | makeProviderServiceLive().pipe( |
| 236 | 242 | Layer.provide(providerAdapterLayer), |
| 237 | 243 | Layer.provide(directoryLayer), |
| 244 | Layer.provideMerge(noopAnalyticsLayer), | |
| 238 | 245 | ), |
| 239 | 246 | directoryLayer, |
| 247 | ||
| 240 | 248 | runtimeRepositoryLayer, |
| 241 | 249 | NodeServices.layer, |
| 242 | 250 | ), |
| ... | ... | @@ -280,6 +288,7 @@ it.effect("ProviderServiceLive keeps persisted resumable sessions on startup", ( |
| 280 | 288 | const providerLayer = makeProviderServiceLive().pipe( |
| 281 | 289 | Layer.provide(Layer.succeed(ProviderAdapterRegistry, registry)), |
| 282 | 290 | Layer.provide(directoryLayer), |
| 291 | Layer.provide(noopAnalyticsLayer), | |
| 283 | 292 | ); |
| 284 | 293 | |
| 285 | 294 | yield* Effect.gen(function* () { |
| ... | ... | @@ -338,6 +347,7 @@ it.effect( |
| 338 | 347 | const firstProviderLayer = makeProviderServiceLive().pipe( |
| 339 | 348 | Layer.provide(Layer.succeed(ProviderAdapterRegistry, firstRegistry)), |
| 340 | 349 | Layer.provide(firstDirectoryLayer), |
| 350 | Layer.provide(noopAnalyticsLayer), | |
| 341 | 351 | ); |
| 342 | 352 | |
| 343 | 353 | const startedSession = yield* Effect.gen(function* () { |
| ... | ... | @@ -380,6 +390,7 @@ it.effect( |
| 380 | 390 | const secondProviderLayer = makeProviderServiceLive().pipe( |
| 381 | 391 | Layer.provide(Layer.succeed(ProviderAdapterRegistry, secondRegistry)), |
| 382 | 392 | Layer.provide(secondDirectoryLayer), |
| 393 | Layer.provide(noopAnalyticsLayer), | |
| 383 | 394 | ); |
| 384 | 395 | |
| 385 | 396 | secondCodex.startSession.mockClear(); |
apps/server/src/provider/Layers/ProviderService.ts+47| ... | ... | @@ -31,6 +31,7 @@ import { |
| 31 | 31 | type ProviderRuntimeBinding, |
| 32 | 32 | } from "../Services/ProviderSessionDirectory.ts"; |
| 33 | 33 | import { type EventNdjsonLogger, makeEventNdjsonLogger } from "./EventNdjsonLogger.ts"; |
| 34 | import { AnalyticsService } from "../../telemetry/Services/AnalyticsService.ts"; | |
| 34 | 35 | |
| 35 | 36 | export interface ProviderServiceLiveOptions { |
| 36 | 37 | readonly canonicalEventLogPath?: string; |
| ... | ... | @@ -108,6 +109,7 @@ function readPersistedCwd( |
| 108 | 109 | |
| 109 | 110 | const makeProviderService = (options?: ProviderServiceLiveOptions) => |
| 110 | 111 | Effect.gen(function* () { |
| 112 | const analytics = yield* Effect.service(AnalyticsService); | |
| 111 | 113 | const canonicalEventLogger = |
| 112 | 114 | options?.canonicalEventLogger ?? |
| 113 | 115 | (options?.canonicalEventLogPath !== undefined |
| ... | ... | @@ -178,6 +180,11 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) => |
| 178 | 180 | const existing = activeSessions.find((session) => session.threadId === input.binding.threadId); |
| 179 | 181 | if (existing) { |
| 180 | 182 | yield* upsertSessionBinding(existing, input.binding.threadId); |
| 183 | yield* analytics.record("provider.session.recovered", { | |
| 184 | provider: existing.provider, | |
| 185 | strategy: "adopt-existing", | |
| 186 | hasResumeCursor: existing.resumeCursor !== undefined, | |
| 187 | }); | |
| 181 | 188 | return { adapter, session: existing } as const; |
| 182 | 189 | } |
| 183 | 190 | } |
| ... | ... | @@ -206,6 +213,11 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) => |
| 206 | 213 | } |
| 207 | 214 | |
| 208 | 215 | yield* upsertSessionBinding(resumed, input.binding.threadId); |
| 216 | yield* analytics.record("provider.session.recovered", { | |
| 217 | provider: resumed.provider, | |
| 218 | strategy: "resume-thread", | |
| 219 | hasResumeCursor: resumed.resumeCursor !== undefined, | |
| 220 | }); | |
| 209 | 221 | return { adapter, session: resumed } as const; |
| 210 | 222 | }); |
| 211 | 223 | |
| ... | ... | @@ -262,6 +274,13 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) => |
| 262 | 274 | } |
| 263 | 275 | |
| 264 | 276 | yield* upsertSessionBinding(session, threadId); |
| 277 | yield* analytics.record("provider.session.started", { | |
| 278 | provider: session.provider, | |
| 279 | hasResumeCursor: session.resumeCursor !== undefined, | |
| 280 | hasCwd: typeof input.cwd === "string" && input.cwd.trim().length > 0, | |
| 281 | approvalPolicy: input.approvalPolicy, | |
| 282 | sandboxMode: input.sandboxMode, | |
| 283 | }); | |
| 265 | 284 | |
| 266 | 285 | return session; |
| 267 | 286 | }); |
| ... | ... | @@ -301,6 +320,11 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) => |
| 301 | 320 | lastRuntimeEventAt: new Date().toISOString(), |
| 302 | 321 | }, |
| 303 | 322 | }); |
| 323 | yield* analytics.record("provider.turn.sent", { | |
| 324 | provider: routed.adapter.provider, | |
| 325 | attachmentCount: input.attachments.length, | |
| 326 | hasInput: typeof input.input === "string" && input.input.trim().length > 0, | |
| 327 | }); | |
| 304 | 328 | return turn; |
| 305 | 329 | }); |
| 306 | 330 | |
| ... | ... | @@ -317,6 +341,10 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) => |
| 317 | 341 | allowRecovery: true, |
| 318 | 342 | }); |
| 319 | 343 | yield* routed.adapter.interruptTurn(routed.threadId, input.turnId); |
| 344 | yield* analytics.record("provider.turn.interrupted", { | |
| 345 | provider: routed.adapter.provider, | |
| 346 | hasTurnId: input.turnId !== undefined, | |
| 347 | }); | |
| 320 | 348 | }); |
| 321 | 349 | |
| 322 | 350 | const respondToRequest: ProviderServiceShape["respondToRequest"] = (rawInput) => |
| ... | ... | @@ -332,6 +360,10 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) => |
| 332 | 360 | allowRecovery: true, |
| 333 | 361 | }); |
| 334 | 362 | yield* routed.adapter.respondToRequest(routed.threadId, input.requestId, input.decision); |
| 363 | yield* analytics.record("provider.request.responded", { | |
| 364 | provider: routed.adapter.provider, | |
| 365 | decision: input.decision, | |
| 366 | }); | |
| 335 | 367 | }); |
| 336 | 368 | |
| 337 | 369 | const respondToUserInput: ProviderServiceShape["respondToUserInput"] = (rawInput) => |
| ... | ... | @@ -347,6 +379,9 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) => |
| 347 | 379 | allowRecovery: true, |
| 348 | 380 | }); |
| 349 | 381 | yield* routed.adapter.respondToUserInput(routed.threadId, input.requestId, input.answers); |
| 382 | yield* analytics.record("provider.user_input.responded", { | |
| 383 | provider: routed.adapter.provider, | |
| 384 | }); | |
| 350 | 385 | }); |
| 351 | 386 | |
| 352 | 387 | const stopSession: ProviderServiceShape["stopSession"] = (rawInput) => |
| ... | ... | @@ -365,6 +400,10 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) => |
| 365 | 400 | yield* routed.adapter.stopSession(routed.threadId); |
| 366 | 401 | } |
| 367 | 402 | yield* directory.remove(input.threadId); |
| 403 | yield* analytics.record("provider.session.stopped", { | |
| 404 | provider: routed.adapter.provider, | |
| 405 | wasActive: routed.isActive, | |
| 406 | }); | |
| 368 | 407 | }); |
| 369 | 408 | |
| 370 | 409 | const listSessions: ProviderServiceShape["listSessions"] = () => |
| ... | ... | @@ -429,6 +468,10 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) => |
| 429 | 468 | allowRecovery: true, |
| 430 | 469 | }); |
| 431 | 470 | yield* routed.adapter.rollbackThread(routed.threadId, input.numTurns); |
| 471 | yield* analytics.record("provider.conversation.rolled_back", { | |
| 472 | provider: routed.adapter.provider, | |
| 473 | turns: input.numTurns, | |
| 474 | }); | |
| 432 | 475 | }); |
| 433 | 476 | |
| 434 | 477 | const stopAll: ProviderServiceShape["stopAll"] = () => |
| ... | ... | @@ -451,6 +494,10 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) => |
| 451 | 494 | ), |
| 452 | 495 | ), |
| 453 | 496 | ).pipe(Effect.asVoid); |
| 497 | yield* analytics.record("provider.sessions.stopped_all", { | |
| 498 | sessionCount: threadIds.length, | |
| 499 | }); | |
| 500 | yield* analytics.flush; | |
| 454 | 501 | }); |
| 455 | 502 | |
| 456 | 503 | return { |
apps/server/src/serverLayers.ts+2-1| ... | ... | @@ -34,11 +34,12 @@ import { CodexTextGenerationLive } from "./git/Layers/CodexTextGeneration"; |
| 34 | 34 | import { GitServiceLive } from "./git/Layers/GitService"; |
| 35 | 35 | import { BunPtyAdapterLive } from "./terminal/Layers/BunPTY"; |
| 36 | 36 | import { NodePtyAdapterLive } from "./terminal/Layers/NodePTY"; |
| 37 | import { AnalyticsService } from "./telemetry/Services/AnalyticsService"; | |
| 37 | 38 | |
| 38 | 39 | export function makeServerProviderLayer(): Layer.Layer< |
| 39 | 40 | ProviderService, |
| 40 | 41 | ProviderUnsupportedError, |
| 41 | SqlClient.SqlClient | ServerConfig | FileSystem.FileSystem | |
| 42 | SqlClient.SqlClient | ServerConfig | FileSystem.FileSystem | AnalyticsService | |
| 42 | 43 | > { |
| 43 | 44 | return Effect.gen(function* () { |
| 44 | 45 | const { stateDir } = yield* ServerConfig; |
apps/server/src/telemetry/Identify.ts created+77| ... | ... | @@ -0,0 +1,77 @@ |
| 1 | import { Effect, FileSystem, Path, Random, Schema } from "effect"; | |
| 2 | import * as Crypto from "node:crypto"; | |
| 3 | import { homedir } from "node:os"; | |
| 4 | ||
| 5 | const CodexAuthJsonSchema = Schema.Struct({ | |
| 6 | tokens: Schema.Struct({ | |
| 7 | account_id: Schema.String, | |
| 8 | }), | |
| 9 | }); | |
| 10 | ||
| 11 | class IdentifyUserError extends Schema.TaggedErrorClass<IdentifyUserError>()("IdentifyUserError", { | |
| 12 | message: Schema.String, | |
| 13 | cause: Schema.optional(Schema.Defect), | |
| 14 | }) {} | |
| 15 | ||
| 16 | const hash = (value: string) => | |
| 17 | Effect.try({ | |
| 18 | try: () => Crypto.createHash("sha256").update(value).digest("hex"), | |
| 19 | catch: (error) => | |
| 20 | new IdentifyUserError({ | |
| 21 | message: "Failed to hash identifier", | |
| 22 | cause: error, | |
| 23 | }), | |
| 24 | }); | |
| 25 | ||
| 26 | const getCodexAccountId = Effect.gen(function* () { | |
| 27 | const fileSystem = yield* FileSystem.FileSystem; | |
| 28 | const path = yield* Path.Path; | |
| 29 | ||
| 30 | const authJsonPath = path.join(homedir(), ".codex", "auth.json"); | |
| 31 | const authJson = yield* Effect.flatMap( | |
| 32 | fileSystem.readFileString(authJsonPath), | |
| 33 | Schema.decodeEffect(Schema.fromJsonString(CodexAuthJsonSchema)), | |
| 34 | ); | |
| 35 | ||
| 36 | return authJson.tokens.account_id; | |
| 37 | }); | |
| 38 | ||
| 39 | const upsertAnonymousId = Effect.gen(function* () { | |
| 40 | const fileSystem = yield* FileSystem.FileSystem; | |
| 41 | const path = yield* Path.Path; | |
| 42 | ||
| 43 | const anonymousIdPath = path.join(homedir(), ".t3", "telemetry", "anonymous-id"); | |
| 44 | const anonymousId = yield* fileSystem.readFileString(anonymousIdPath).pipe( | |
| 45 | Effect.catch(() => | |
| 46 | Effect.gen(function* () { | |
| 47 | const randomId = yield* Random.nextUUIDv4; | |
| 48 | yield* fileSystem.writeFileString(anonymousIdPath, randomId); | |
| 49 | return randomId; | |
| 50 | }), | |
| 51 | ), | |
| 52 | ); | |
| 53 | ||
| 54 | return anonymousId; | |
| 55 | }); | |
| 56 | ||
| 57 | /** | |
| 58 | * getTelemetryIdentifier - Users are "identified" by finding the first match of the following, then hashing the value. | |
| 59 | * 1. ~/.codex/auth.json tokens.account_id | |
| 60 | * 2. ~/.t3/telemetry/anonymous-id | |
| 61 | */ | |
| 62 | export const getTelemetryIdentifier = Effect.gen(function* () { | |
| 63 | const codexAccountId = yield* getCodexAccountId; | |
| 64 | if (codexAccountId) { | |
| 65 | return yield* hash(codexAccountId); | |
| 66 | } | |
| 67 | ||
| 68 | const anonymousId = yield* upsertAnonymousId; | |
| 69 | if (anonymousId) { | |
| 70 | return yield* hash(anonymousId); | |
| 71 | } | |
| 72 | ||
| 73 | return null; | |
| 74 | }).pipe( | |
| 75 | Effect.tapError((error) => Effect.logWarning("Failed to get identifier", { cause: error })), | |
| 76 | Effect.orElseSucceed(() => null), | |
| 77 | ); |
apps/server/src/telemetry/Layers/AnalyticsService.ts created+137| ... | ... | @@ -0,0 +1,137 @@ |
| 1 | /** | |
| 2 | * AnalyticsServiceLive - Anonymous PostHog telemetry layer. | |
| 3 | * | |
| 4 | * Persists a random installation-scoped anonymous id to state dir, buffers | |
| 5 | * events in memory, and flushes batches to PostHog over Effect HttpClient. | |
| 6 | * | |
| 7 | * @module AnalyticsServiceLive | |
| 8 | */ | |
| 9 | ||
| 10 | import { Config, DateTime, Effect, Layer, Ref } from "effect"; | |
| 11 | import { HttpClient, HttpClientRequest, HttpClientResponse } from "effect/unstable/http"; | |
| 12 | ||
| 13 | import { AnalyticsService, type AnalyticsServiceShape } from "../Services/AnalyticsService.ts"; | |
| 14 | import { getTelemetryIdentifier } from "../Identify.ts"; | |
| 15 | ||
| 16 | interface BufferedAnalyticsEvent { | |
| 17 | readonly event: string; | |
| 18 | readonly properties?: Readonly<Record<string, unknown>>; | |
| 19 | readonly capturedAt: string; | |
| 20 | } | |
| 21 | ||
| 22 | const TelemetryEnvConfig = Config.all({ | |
| 23 | posthogKey: Config.string("T3CODE_POSTHOG_KEY").pipe( | |
| 24 | Config.withDefault("phc_XOWci4oZP4VvLiEyrFqkFjP4CZn55mjYYBMREK5Wd6m"), | |
| 25 | ), | |
| 26 | posthogHost: Config.string("T3CODE_POSTHOG_HOST").pipe( | |
| 27 | Config.withDefault("https://us.i.posthog.com"), | |
| 28 | ), | |
| 29 | enabled: Config.boolean("T3CODE_TELEMETRY_ENABLED").pipe(Config.withDefault(true)), | |
| 30 | flushBatchSize: Config.number("T3CODE_TELEMETRY_FLUSH_BATCH_SIZE").pipe(Config.withDefault(20)), | |
| 31 | maxBufferedEvents: Config.number("T3CODE_TELEMETRY_MAX_BUFFERED_EVENTS").pipe( | |
| 32 | Config.withDefault(1_000), | |
| 33 | ), | |
| 34 | }); | |
| 35 | ||
| 36 | const makeAnalyticsService = Effect.gen(function* () { | |
| 37 | const telemetryConfig = yield* TelemetryEnvConfig.asEffect(); | |
| 38 | const httpClient = yield* HttpClient.HttpClient; | |
| 39 | const identifier = yield* getTelemetryIdentifier; | |
| 40 | const bufferRef = yield* Ref.make<ReadonlyArray<BufferedAnalyticsEvent>>([]); | |
| 41 | ||
| 42 | const enqueueBufferedEvent = (event: string, properties?: Readonly<Record<string, unknown>>) => | |
| 43 | Effect.flatMap(DateTime.now, (now) => | |
| 44 | Ref.modify(bufferRef, (current) => { | |
| 45 | const appended = [ | |
| 46 | ...current, | |
| 47 | { | |
| 48 | event, | |
| 49 | ...(properties ? { properties } : {}), | |
| 50 | capturedAt: DateTime.formatIso(now), | |
| 51 | } satisfies BufferedAnalyticsEvent, | |
| 52 | ]; | |
| 53 | ||
| 54 | const next = | |
| 55 | appended.length > telemetryConfig.maxBufferedEvents | |
| 56 | ? appended.slice(appended.length - telemetryConfig.maxBufferedEvents) | |
| 57 | : appended; | |
| 58 | ||
| 59 | return [ | |
| 60 | { | |
| 61 | size: next.length, | |
| 62 | dropped: next.length !== appended.length, | |
| 63 | } as const, | |
| 64 | next, | |
| 65 | ] as const; | |
| 66 | }), | |
| 67 | ); | |
| 68 | ||
| 69 | const sendBatch = (events: ReadonlyArray<BufferedAnalyticsEvent>) => | |
| 70 | Effect.gen(function* () { | |
| 71 | if (!telemetryConfig.enabled || !identifier) return; | |
| 72 | ||
| 73 | const payload = { | |
| 74 | api_key: telemetryConfig.posthogKey, | |
| 75 | batch: events.map((event) => ({ | |
| 76 | event: event.event, | |
| 77 | distinct_id: identifier, | |
| 78 | properties: { | |
| 79 | ...event.properties, | |
| 80 | $process_person_profile: false, | |
| 81 | $lib: "t3code-server", | |
| 82 | }, | |
| 83 | timestamp: event.capturedAt, | |
| 84 | })), | |
| 85 | }; | |
| 86 | ||
| 87 | yield* HttpClientRequest.post(`${telemetryConfig.posthogHost}/batch/`).pipe( | |
| 88 | HttpClientRequest.bodyJson(payload), | |
| 89 | Effect.flatMap(httpClient.execute), | |
| 90 | Effect.flatMap(HttpClientResponse.filterStatusOk), | |
| 91 | ); | |
| 92 | }); | |
| 93 | ||
| 94 | const flush: AnalyticsServiceShape["flush"] = Effect.gen(function* () { | |
| 95 | const batch = yield* Ref.modify(bufferRef, (current) => { | |
| 96 | if (current.length === 0) { | |
| 97 | return [[] as ReadonlyArray<BufferedAnalyticsEvent>, current] as const; | |
| 98 | } | |
| 99 | const nextBatch = current.slice(0, telemetryConfig.flushBatchSize); | |
| 100 | const remaining = current.slice(nextBatch.length); | |
| 101 | return [nextBatch, remaining] as const; | |
| 102 | }); | |
| 103 | ||
| 104 | if (batch.length === 0) { | |
| 105 | return; | |
| 106 | } | |
| 107 | ||
| 108 | yield* sendBatch(batch).pipe( | |
| 109 | Effect.catch((error) => | |
| 110 | Ref.update(bufferRef, (current) => [...batch, ...current]).pipe( | |
| 111 | Effect.flatMap(() => Effect.fail(error)), | |
| 112 | ), | |
| 113 | ), | |
| 114 | ); | |
| 115 | }).pipe(Effect.catch((cause) => Effect.logError("Failed to flush telemetry", { cause }))); | |
| 116 | ||
| 117 | const record: AnalyticsServiceShape["record"] = Effect.fnUntraced(function* (event, properties) { | |
| 118 | if (!telemetryConfig.enabled || !identifier) return; | |
| 119 | ||
| 120 | const enqueueResult = yield* enqueueBufferedEvent(event, properties); | |
| 121 | if (enqueueResult.dropped) { | |
| 122 | yield* Effect.logDebug("analytics buffer full; dropping oldest event", { | |
| 123 | size: enqueueResult.size, | |
| 124 | event, | |
| 125 | }); | |
| 126 | } | |
| 127 | }); | |
| 128 | ||
| 129 | yield* Effect.forever(flush, { disableYield: true }).pipe(Effect.forkDetach); | |
| 130 | ||
| 131 | return { | |
| 132 | record, | |
| 133 | flush, | |
| 134 | } satisfies AnalyticsServiceShape; | |
| 135 | }); | |
| 136 | ||
| 137 | export const AnalyticsServiceLayerLive = Layer.effect(AnalyticsService, makeAnalyticsService); |
apps/server/src/telemetry/Services/AnalyticsService.ts created+28| ... | ... | @@ -0,0 +1,28 @@ |
| 1 | /** | |
| 2 | * AnalyticsService - Anonymous telemetry capture contract. | |
| 3 | * | |
| 4 | * Provides a best-effort event API for runtime telemetry and a strict | |
| 5 | * `captureImmediate` method for call sites that need explicit error handling. | |
| 6 | * | |
| 7 | * @module AnalyticsService | |
| 8 | */ | |
| 9 | import { Effect, ServiceMap } from "effect"; | |
| 10 | ||
| 11 | export interface AnalyticsServiceShape { | |
| 12 | /** | |
| 13 | * Capture an event immediately; returns typed failure when capture fails. | |
| 14 | */ | |
| 15 | readonly record: ( | |
| 16 | event: string, | |
| 17 | properties?: Readonly<Record<string, unknown>>, | |
| 18 | ) => Effect.Effect<void, never>; | |
| 19 | ||
| 20 | /** | |
| 21 | * Flush queued telemetry. | |
| 22 | */ | |
| 23 | readonly flush: Effect.Effect<void, never>; | |
| 24 | } | |
| 25 | ||
| 26 | export class AnalyticsService extends ServiceMap.Service<AnalyticsService, AnalyticsServiceShape>()( | |
| 27 | "t3/telemetry/Services/AnalyticsService", | |
| 28 | ) {} |
apps/server/src/wsServer.test.ts+5-5| ... | ... | @@ -50,6 +50,8 @@ import type { GitCoreShape } from "./git/Services/GitCore.ts"; |
| 50 | 50 | import { GitCore } from "./git/Services/GitCore.ts"; |
| 51 | 51 | import { GitCommandError, GitManagerError } from "./git/Errors.ts"; |
| 52 | 52 | import { MigrationError } from "@effect/sql-sqlite-bun/SqliteMigrator"; |
| 53 | import { AnalyticsServiceLayerLive } from "./telemetry/Layers/AnalyticsService.ts"; | |
| 54 | import * as NodeHttpClient from "@effect/platform-node/NodeHttpClient"; | |
| 53 | 55 | |
| 54 | 56 | interface PendingMessages { |
| 55 | 57 | queue: unknown[]; |
| ... | ... | @@ -393,10 +395,7 @@ describe("WebSocket Server", () => { |
| 393 | 395 | providerHealth?: ProviderHealthShape; |
| 394 | 396 | open?: OpenShape; |
| 395 | 397 | gitManager?: GitManagerShape; |
| 396 | gitCore?: Pick< | |
| 397 | GitCoreShape, | |
| 398 | "listBranches" | "initRepo" | "pullCurrentBranch" | |
| 399 | >; | |
| 398 | gitCore?: Pick<GitCoreShape, "listBranches" | "initRepo" | "pullCurrentBranch">; | |
| 400 | 399 | terminalManager?: TerminalManagerShape; |
| 401 | 400 | } = {}, |
| 402 | 401 | ): Promise<Http.Server> { |
| ... | ... | @@ -449,7 +448,9 @@ describe("WebSocket Server", () => { |
| 449 | 448 | Layer.provideMerge(providerHealthLayer), |
| 450 | 449 | Layer.provideMerge(openLayer), |
| 451 | 450 | Layer.provideMerge(serverConfigLayer), |
| 451 | Layer.provideMerge(AnalyticsServiceLayerLive), | |
| 452 | 452 | Layer.provideMerge(NodeServices.layer), |
| 453 | Layer.provideMerge(NodeHttpClient.layerUndici), | |
| 453 | 454 | ); |
| 454 | 455 | const runtimeServices = await Effect.runPromise( |
| 455 | 456 | Layer.build(dependenciesLayer).pipe(Scope.provide(scope)), |
| ... | ... | @@ -1620,7 +1621,6 @@ describe("WebSocket Server", () => { |
| 1620 | 1621 | expect(pullResponse.result).toBeUndefined(); |
| 1621 | 1622 | expect(pullResponse.error?.message).toContain("No upstream configured"); |
| 1622 | 1623 | expect(pullCurrentBranch).toHaveBeenCalledWith("/repo/path"); |
| 1623 | ||
| 1624 | 1624 | }); |
| 1625 | 1625 | |
| 1626 | 1626 | it("supports git.status over websocket", async () => { |
apps/server/src/wsServer.ts+3-1| ... | ... | @@ -71,6 +71,7 @@ import { |
| 71 | 71 | resolveAttachmentPathById, |
| 72 | 72 | } from "./attachmentStore.ts"; |
| 73 | 73 | import { parseBase64DataUrl } from "./imageMime.ts"; |
| 74 | import { AnalyticsService } from "./telemetry/Services/AnalyticsService.ts"; | |
| 74 | 75 | |
| 75 | 76 | /** |
| 76 | 77 | * ServerShape - Service API for server lifecycle control. |
| ... | ... | @@ -207,7 +208,8 @@ export type ServerRuntimeServices = |
| 207 | 208 | | GitCore |
| 208 | 209 | | TerminalManager |
| 209 | 210 | | Keybindings |
| 210 | | Open; | |
| 211 | | Open | |
| 212 | | AnalyticsService; | |
| 211 | 213 | |
| 212 | 214 | export class ServerLifecycleError extends Schema.TaggedErrorClass<ServerLifecycleError>()( |
| 213 | 215 | "ServerLifecycleError", |