| ... | ... | @@ -0,0 +1,121 @@ |
| 1 | import fs from "node:fs"; |
| 2 | import os from "node:os"; |
| 3 | import path from "node:path"; |
| 4 | |
| 5 | import * as NodeHttpServer from "@effect/platform-node/NodeHttpServer"; |
| 6 | import { assert, it } from "@effect/vitest"; |
| 7 | import { ConfigProvider, Effect, Layer } from "effect"; |
| 8 | import * as HttpServer from "effect/unstable/http/HttpServer"; |
| 9 | import * as HttpServerRequest from "effect/unstable/http/HttpServerRequest"; |
| 10 | import * as HttpServerResponse from "effect/unstable/http/HttpServerResponse"; |
| 11 | |
| 12 | import { ServerConfig } from "../../config.ts"; |
| 13 | import { getTelemetryIdentifier } from "../Identify.ts"; |
| 14 | import { AnalyticsService } from "../Services/AnalyticsService.ts"; |
| 15 | import { AnalyticsServiceLayerLive } from "./AnalyticsService.ts"; |
| 16 | |
| 17 | interface RecordedBatchRequest { |
| 18 | readonly path: string; |
| 19 | readonly body: |
| 20 | | { |
| 21 | readonly batch?: ReadonlyArray<{ |
| 22 | readonly event?: string; |
| 23 | readonly properties?: { |
| 24 | readonly index?: number; |
| 25 | }; |
| 26 | }>; |
| 27 | } |
| 28 | | null; |
| 29 | } |
| 30 | |
| 31 | interface RecordedBatchBody { |
| 32 | readonly batch: ReadonlyArray<{ |
| 33 | readonly event?: string; |
| 34 | readonly properties?: { |
| 35 | readonly index?: number; |
| 36 | }; |
| 37 | }>; |
| 38 | } |
| 39 | |
| 40 | it.effect("flush drains all buffered events across multiple batches", () => |
| 41 | Effect.gen(function* () { |
| 42 | const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-telemetry-flush-")); |
| 43 | const capturedRequests: Array<RecordedBatchRequest> = []; |
| 44 | const telemetryLayer = AnalyticsServiceLayerLive.pipe( |
| 45 | Layer.provideMerge(ServerConfig.layerTest(process.cwd(), tempDir)), |
| 46 | ); |
| 47 | const configLayer = ConfigProvider.layer( |
| 48 | ConfigProvider.fromUnknown({ |
| 49 | T3CODE_TELEMETRY_ENABLED: true, |
| 50 | T3CODE_POSTHOG_KEY: "phc_test_key", |
| 51 | T3CODE_POSTHOG_HOST: "", |
| 52 | T3CODE_TELEMETRY_FLUSH_BATCH_SIZE: 20, |
| 53 | }), |
| 54 | ); |
| 55 | const batchServerLayer = HttpServer.serve( |
| 56 | Effect.gen(function* () { |
| 57 | const request = yield* HttpServerRequest.HttpServerRequest; |
| 58 | if (request.method !== "POST") { |
| 59 | return HttpServerResponse.empty({ status: 404 }); |
| 60 | } |
| 61 | |
| 62 | const payload = yield* request.json.pipe( |
| 63 | Effect.map((body) => body as RecordedBatchRequest["body"]), |
| 64 | Effect.catch(() => Effect.succeed(null)), |
| 65 | ); |
| 66 | yield* Effect.sync(() => { |
| 67 | capturedRequests.push({ path: request.url, body: payload }); |
| 68 | }); |
| 69 | |
| 70 | return HttpServerResponse.jsonUnsafe({}); |
| 71 | }), |
| 72 | ); |
| 73 | const runtimeLayer = telemetryLayer.pipe( |
| 74 | Layer.provide(configLayer), |
| 75 | Layer.provideMerge(NodeHttpServer.layerTest), |
| 76 | ); |
| 77 | |
| 78 | yield* Effect.gen(function* () { |
| 79 | yield* Layer.launch(batchServerLayer).pipe(Effect.forkScoped); |
| 80 | const telemetryIdentifier = yield* getTelemetryIdentifier; |
| 81 | assert.equal(telemetryIdentifier !== null, true); |
| 82 | const analytics = yield* AnalyticsService; |
| 83 | |
| 84 | for (let index = 0; index < 45; index += 1) { |
| 85 | yield* analytics.record("test.flush.drain", { index }); |
| 86 | } |
| 87 | |
| 88 | yield* analytics.flush; |
| 89 | }).pipe( |
| 90 | Effect.provide(runtimeLayer), |
| 91 | Effect.ensuring( |
| 92 | Effect.sync(() => { |
| 93 | fs.rmSync(tempDir, { recursive: true, force: true }); |
| 94 | }), |
| 95 | ), |
| 96 | ); |
| 97 | |
| 98 | const batchRequests = capturedRequests.filter( |
| 99 | (request): request is RecordedBatchRequest & { readonly body: RecordedBatchBody } => |
| 100 | Array.isArray(request.body?.batch), |
| 101 | ); |
| 102 | assert.equal(batchRequests.length, 3); |
| 103 | assert.equal( |
| 104 | batchRequests.every((request) => request.path === "/batch/" || request.path === "/batch"), |
| 105 | true, |
| 106 | ); |
| 107 | const deliveredIndexes = batchRequests.flatMap((request) => |
| 108 | request.body.batch |
| 109 | .filter((event) => event.event === "test.flush.drain") |
| 110 | .map((event) => event.properties?.index) |
| 111 | .filter((index): index is number => typeof index === "number"), |
| 112 | ); |
| 113 | |
| 114 | const sorted = deliveredIndexes.toSorted((a, b) => a - b); |
| 115 | assert.equal(sorted.length, 45); |
| 116 | assert.deepEqual( |
| 117 | sorted, |
| 118 | Array.from({ length: 45 }, (_, index) => index), |
| 119 | ); |
| 120 | }), |
| 121 | ); |