import * as NodeHttpServer from "@effect/platform-node/NodeHttpServer";import * as NodeServices from "@effect/platform-node/NodeServices";import { assert, it } from "@effect/vitest";import * as ConfigProvider from "effect/ConfigProvider";import * as Effect from "effect/Effect";import * as Layer from "effect/Layer";import * as HttpServer from "effect/unstable/http/HttpServer";import * as HttpServerRequest from "effect/unstable/http/HttpServerRequest";import * as HttpServerResponse from "effect/unstable/http/HttpServerResponse"; import * as ServerConfig from "../config.ts";import { getTelemetryIdentifier } from "./Identify.ts";import * as AnalyticsService from "./AnalyticsService.ts"; interface RecordedBatchRequest { readonly path: string; readonly body: { readonly batch?: ReadonlyArray<{ readonly event?: string; readonly properties?: { readonly index?: number; readonly clientType?: string; }; }>; } | null;} interface RecordedBatchBody { readonly batch: ReadonlyArray<{ readonly event?: string; readonly properties?: { readonly index?: number; readonly clientType?: string; }; }>;} it.layer(NodeServices.layer)("AnalyticsService test", (it) => { it.effect("flush drains all buffered events across multiple batches", () => Effect.gen(function* () { const capturedRequests: Array<RecordedBatchRequest> = []; const serverConfigLayer = ServerConfig.ServerConfig.layerTest(process.cwd(), { prefix: "t3-telemetry-base-", }); const telemetryLayer = AnalyticsService.layer.pipe(Layer.provideMerge(serverConfigLayer)); const configLayer = ConfigProvider.layer( ConfigProvider.fromUnknown({ T3CODE_TELEMETRY_ENABLED: true, T3CODE_POSTHOG_KEY: "phc_test_key", T3CODE_POSTHOG_HOST: "", T3CODE_TELEMETRY_FLUSH_BATCH_SIZE: 20, }), ); const batchServerLayer = HttpServer.serve( Effect.gen(function* () { const request = yield* HttpServerRequest.HttpServerRequest; if (request.method !== "POST") { return HttpServerResponse.empty({ status: 404 }); } const payload = yield* request.json.pipe( Effect.map((body) => body as RecordedBatchRequest["body"]), Effect.orElseSucceed(() => null), ); capturedRequests.push({ path: request.url, body: payload }); return HttpServerResponse.jsonUnsafe({}); }), ); const runtimeLayer = telemetryLayer.pipe( Layer.provide(configLayer), Layer.provideMerge(NodeHttpServer.layerTest), ); yield* Effect.gen(function* () { yield* Layer.launch(batchServerLayer).pipe(Effect.forkScoped); const telemetryIdentifier = yield* getTelemetryIdentifier; assert.equal(telemetryIdentifier !== null, true); const analytics = yield* AnalyticsService.AnalyticsService; for (let index = 0; index < 45; index += 1) { yield* analytics.record("test.flush.drain", { index }); } yield* analytics.flush; }).pipe(Effect.provide(runtimeLayer)); const batchRequests = capturedRequests.filter( (request): request is RecordedBatchRequest & { readonly body: RecordedBatchBody } => Array.isArray(request.body?.batch), ); assert.equal(batchRequests.length, 3); assert.equal( batchRequests.every((request) => request.path === "/batch/" || request.path === "/batch"), true, ); const deliveredIndexes = batchRequests.flatMap((request) => request.body.batch .filter((event) => event.event === "test.flush.drain") .map((event) => event.properties?.index) .filter((index): index is number => typeof index === "number"), ); const sorted = deliveredIndexes.toSorted((a, b) => a - b); assert.equal(sorted.length, 45); assert.deepEqual( sorted, Array.from({ length: 45 }, (_, index) => index), ); assert.equal( batchRequests.every((request) => request.body.batch.every((event) => event.properties?.clientType === "cli-web-client"), ), true, ); }), );});
import * as NodeHttpServer from "@effect/platform-node/NodeHttpServer";
import * as NodeServices from "@effect/platform-node/NodeServices";
import { assert, it } from "@effect/vitest";
import * as ConfigProvider from "effect/ConfigProvider";
import * as Effect from "effect/Effect";
import * as Layer from "effect/Layer";
import * as HttpServer from "effect/unstable/http/HttpServer";
import * as HttpServerRequest from "effect/unstable/http/HttpServerRequest";
import * as HttpServerResponse from "effect/unstable/http/HttpServerResponse";
import * as ServerConfig from "../config.ts";
import { getTelemetryIdentifier } from "./Identify.ts";
import * as AnalyticsService from "./AnalyticsService.ts";
interface RecordedBatchRequest {
readonly path: string;
readonly body: {
readonly batch?: ReadonlyArray<{
readonly event?: string;
readonly properties?: {
readonly index?: number;
readonly clientType?: string;
};
}>;
} | null;
}
interface RecordedBatchBody {
readonly batch: ReadonlyArray<{
readonly event?: string;
readonly properties?: {
readonly index?: number;
readonly clientType?: string;
};
}>;
}
it.layer(NodeServices.layer)("AnalyticsService test", (it) => {
it.effect("flush drains all buffered events across multiple batches", () =>
Effect.gen(function* () {
const capturedRequests: Array<RecordedBatchRequest> = [];
const serverConfigLayer = ServerConfig.ServerConfig.layerTest(process.cwd(), {
prefix: "t3-telemetry-base-",
});
const telemetryLayer = AnalyticsService.layer.pipe(Layer.provideMerge(serverConfigLayer));
const configLayer = ConfigProvider.layer(
ConfigProvider.fromUnknown({
T3CODE_TELEMETRY_ENABLED: true,
T3CODE_POSTHOG_KEY: "phc_test_key",
T3CODE_POSTHOG_HOST: "",
T3CODE_TELEMETRY_FLUSH_BATCH_SIZE: 20,
}),
);
const batchServerLayer = HttpServer.serve(
Effect.gen(function* () {
const request = yield* HttpServerRequest.HttpServerRequest;
if (request.method !== "POST") {
return HttpServerResponse.empty({ status: 404 });
}
const payload = yield* request.json.pipe(
Effect.map((body) => body as RecordedBatchRequest["body"]),
Effect.orElseSucceed(() => null),
);
capturedRequests.push({ path: request.url, body: payload });
return HttpServerResponse.jsonUnsafe({});
}),
);
const runtimeLayer = telemetryLayer.pipe(
Layer.provide(configLayer),
Layer.provideMerge(NodeHttpServer.layerTest),
);
yield* Effect.gen(function* () {
yield* Layer.launch(batchServerLayer).pipe(Effect.forkScoped);
const telemetryIdentifier = yield* getTelemetryIdentifier;
assert.equal(telemetryIdentifier !== null, true);
const analytics = yield* AnalyticsService.AnalyticsService;
for (let index = 0; index < 45; index += 1) {
yield* analytics.record("test.flush.drain", { index });
}
yield* analytics.flush;
}).pipe(Effect.provide(runtimeLayer));
const batchRequests = capturedRequests.filter(
(request): request is RecordedBatchRequest & { readonly body: RecordedBatchBody } =>
Array.isArray(request.body?.batch),
);
assert.equal(batchRequests.length, 3);
assert.equal(
batchRequests.every((request) => request.path === "/batch/" || request.path === "/batch"),
true,
);
const deliveredIndexes = batchRequests.flatMap((request) =>
request.body.batch
.filter((event) => event.event === "test.flush.drain")
.map((event) => event.properties?.index)
.filter((index): index is number => typeof index === "number"),
);
const sorted = deliveredIndexes.toSorted((a, b) => a - b);
assert.equal(sorted.length, 45);
assert.deepEqual(
sorted,
Array.from({ length: 45 }, (_, index) => index),
);
assert.equal(
batchRequests.every((request) =>
request.body.batch.every((event) => event.properties?.clientType === "cli-web-client"),
),
true,
);
}),
);
});