1// @effect-diagnostics nodeBuiltinImport:off
2import * as NodeFS from "node:fs";
3import * as NodeOS from "node:os";
4import * as NodePath from "node:path";
5
6import { it } from "@effect/vitest";
7import * as DateTime from "effect/DateTime";
8import * as Effect from "effect/Effect";
9import * as Option from "effect/Option";
10import * as PubSub from "effect/PubSub";
11import * as Ref from "effect/Ref";
12import { assert, describe, expect } from "vite-plus/test";
13
14import {
15 type DesktopTelemetryReceiverHealth,
16 initialDesktopTelemetryContactAt,
17 isDesktopTelemetryContactStale,
18 recordDesktopTelemetrySampleHealth,
19 requireDesktopTelemetryWriteProgress,
20 resolveDesktopTelemetrySnapshotStaleAfterMs,
21 writeAllToFileDescriptor,
22} from "./DesktopTelemetryReceiver.ts";
23
24describe("DesktopTelemetryReceiver", () => {
25 it("degrades a hello-only stream after the first-sample deadline", () => {
26 expect(isDesktopTelemetryContactStale(Option.some(1_000), 90_999)).toBe(false);
27 expect(isDesktopTelemetryContactStale(Option.some(1_000), 91_000)).toBe(true);
28 expect(isDesktopTelemetryContactStale(Option.none(), 1_000_000)).toBe(false);
29 });
30
31 it("starts the stale deadline as soon as a telemetry descriptor is opened", () => {
32 expect(initialDesktopTelemetryContactAt(7, 1_000)).toEqual(Option.some(1_000));
33 expect(initialDesktopTelemetryContactAt(undefined, 1_000)).toEqual(Option.none());
34 });
35
36 it("keeps the snapshot deadline beyond the configured idle polling interval", () => {
37 expect(resolveDesktopTelemetrySnapshotStaleAfterMs(30_000, 120_000)).toBe(150_000);
38 expect(resolveDesktopTelemetrySnapshotStaleAfterMs(60_000, 600_000)).toBe(630_000);
39 expect(resolveDesktopTelemetrySnapshotStaleAfterMs(1_000, 1_000)).toBe(90_000);
40 });
41
42 it.effect("publishes the latest sample timestamp while health remains healthy", () =>
43 Effect.scoped(
44 Effect.gen(function* () {
45 const initialSample = DateTime.makeUnsafe(1_000);
46 const nextSample = DateTime.makeUnsafe(2_000);
47 const health = yield* Ref.make<DesktopTelemetryReceiverHealth>({
48 status: "healthy",
49 lastSampleAt: Option.some(initialSample),
50 lastError: Option.none<string>(),
51 });
52 const healthChanges = yield* PubSub.sliding<DesktopTelemetryReceiverHealth>(4);
53 const subscription = yield* PubSub.subscribe(healthChanges);
54
55 yield* recordDesktopTelemetrySampleHealth(health, healthChanges, nextSample);
56 const published = yield* PubSub.take(subscription).pipe(Effect.timeout("1 second"));
57
58 expect(DateTime.toEpochMillis(Option.getOrThrow(published.lastSampleAt))).toBe(2_000);
59 }),
60 ),
61 );
62
63 it.effect("writes control messages through the asynchronous descriptor path", () =>
64 Effect.acquireUseRelease(
65 Effect.sync(() => {
66 const directory = NodeFS.mkdtempSync(
67 NodePath.join(NodeOS.tmpdir(), "t3-desktop-telemetry-control-test-"),
68 );
69 const path = NodePath.join(directory, "control.ndjson");
70 return {
71 directory,
72 path,
73 fd: NodeFS.openSync(path, "w"),
74 };
75 }),
76 ({ fd, path }) =>
77 Effect.gen(function* () {
78 const payload = Buffer.from('{"type":"setDiagnosticsDemand","enabled":true}\n');
79 yield* writeAllToFileDescriptor(fd, payload);
80 NodeFS.fsyncSync(fd);
81
82 assert.equal(NodeFS.readFileSync(path, "utf8"), payload.toString("utf8"));
83 }),
84 ({ directory, fd }) =>
85 Effect.sync(() => {
86 NodeFS.closeSync(fd);
87 NodeFS.rmSync(directory, { recursive: true, force: true });
88 }),
89 ),
90 );
91
92 it.effect("models a zero-byte control write as a stalled descriptor", () =>
93 Effect.gen(function* () {
94 const error = yield* requireDesktopTelemetryWriteProgress(7, 42, 0).pipe(Effect.flip);
95
96 expect(error._tag).toBe("DesktopTelemetryControlStalled");
97 expect(error.fd).toBe(7);
98 expect(error.remainingBytes).toBe(42);
99 expect(error.message).toBe(
100 "Desktop telemetry control stalled on fd 7 with 42 bytes remaining.",
101 );
102 yield* requireDesktopTelemetryWriteProgress(7, 42, 1);
103 }),
104 );
105});