1// @effect-diagnostics nodeBuiltinImport:off
2import * as NodeFS from "node:fs";
3import * as NodeNet from "node:net";
4import * as NodeTls from "node:tls";
5
6import * as NodeStream from "@effect/platform-node/NodeStream";
7import {
8 DesktopHostTelemetryMessage,
9 type DesktopHostTelemetryMessage as DesktopHostTelemetryMessageValue,
10 type DesktopHostTelemetrySnapshot,
11 type DesktopShellEnvironmentSnapshot,
12 DesktopTelemetryControlMessage,
13 type DesktopUpdateChannel,
14 type DesktopUpdateStatusReport,
15 type ResourceTelemetrySourceStatus,
16} from "@t3tools/contracts";
17import { resolveServerBackgroundActivitySettings } from "@t3tools/shared/backgroundActivitySettings";
18import * as HostProcess from "@t3tools/shared/HostProcess";
19import { applyShellEnvironment } from "@t3tools/shared/shell";
20import * as Cause from "effect/Cause";
21import * as Context from "effect/Context";
22import * as DateTime from "effect/DateTime";
23import * as Duration from "effect/Duration";
24import * as Effect from "effect/Effect";
25import * as FileSystem from "effect/FileSystem";
26import * as Layer from "effect/Layer";
27import * as Option from "effect/Option";
28import * as PubSub from "effect/PubSub";
29import * as Ref from "effect/Ref";
30import * as Schema from "effect/Schema";
31import * as Semaphore from "effect/Semaphore";
32import * as Scope from "effect/Scope";
33import * as Stream from "effect/Stream";
34import * as Ndjson from "effect/encoding/Ndjson";
35
36import { eventVersion as messageVersion } from "./NativeTelemetryClient.ts";
37import * as ServerConfig from "../config.ts";
38import * as ServerSettings from "../serverSettings.ts";
39import { subscribeBeforeSnapshotWithoutMutex } from "../utils/subscribeBeforeSnapshot.ts";
40
41const INITIAL_SAMPLE_DEADLINE_MS = 90_000;
42const MIN_SNAPSHOT_STALE_AFTER_MS = 90_000;
43const STALE_GRACE_MS = 30_000;
44const DEFAULT_HOST_POWER_ACTIVE_INTERVAL_MS = 30_000;
45const DEFAULT_HOST_POWER_IDLE_INTERVAL_MS = 120_000;
46const STALE_CHECK_INTERVAL = Duration.seconds(30);
47
48export class DesktopTelemetryDescriptorUnavailable extends Schema.TaggedError<DesktopTelemetryDescriptorUnavailable>()(
49 "DesktopTelemetryDescriptorUnavailable",
50 {
51 mode: Schema.String,
52 },
53) {
54 override get message(): string {
55 return `Desktop telemetry descriptor is unavailable in '${this.mode}' mode.`;
56 }
57}
58
59export class DesktopTelemetryProtocolMismatch extends Schema.TaggedError<DesktopTelemetryProtocolMismatch>()(
60 "DesktopTelemetryProtocolMismatch",
61 {
62 expectedVersion: Schema.Number,
63 receivedVersion: Schema.Number,
64 },
65) {
66 override get message(): string {
67 return `Desktop telemetry protocol ${this.receivedVersion} is incompatible with expected protocol ${this.expectedVersion}.`;
68 }
69}
70
71export class DesktopTelemetryDecodeFailed extends Schema.TaggedError<DesktopTelemetryDecodeFailed>()(
72 "DesktopTelemetryDecodeFailed",
73 {
74 cause: Schema.Defect(),
75 },
76) {
77 override get message(): string {
78 return "Failed to decode desktop telemetry.";
79 }
80}
81
82export class DesktopTelemetryStreamFailed extends Schema.TaggedError<DesktopTelemetryStreamFailed>()(
83 "DesktopTelemetryStreamFailed",
84 {
85 fd: Schema.Number,
86 cause: Schema.Defect(),
87 },
88) {
89 override get message(): string {
90 return `Desktop telemetry stream on fd ${this.fd} failed.`;
91 }
92}
93
94export class DesktopTelemetryStreamClosed extends Schema.TaggedError<DesktopTelemetryStreamClosed>()(
95 "DesktopTelemetryStreamClosed",
96 {
97 fd: Schema.Number,
98 },
99) {
100 override get message(): string {
101 return `Desktop telemetry stream on fd ${this.fd} closed.`;
102 }
103}
104
105export class DesktopTelemetryStale extends Schema.TaggedError<DesktopTelemetryStale>()(
106 "DesktopTelemetryStale",
107 {
108 fd: Schema.Number,
109 staleAfterMs: Schema.Number,
110 },
111) {
112 override get message(): string {
113 return `Desktop telemetry on fd ${this.fd} has not updated for ${this.staleAfterMs}ms.`;
114 }
115}
116
117export type DesktopTelemetryReceiverError =
118 | DesktopTelemetryDescriptorUnavailable
119 | DesktopTelemetryProtocolMismatch
120 | DesktopTelemetryDecodeFailed
121 | DesktopTelemetryStreamFailed
122 | DesktopTelemetryStreamClosed;
123
124export class DesktopTelemetryControlFailed extends Schema.TaggedError<DesktopTelemetryControlFailed>()(
125 "DesktopTelemetryControlFailed",
126 {
127 fd: Schema.Number,
128 operation: Schema.String,
129 cause: Schema.Defect(),
130 },
131) {
132 override get message(): string {
133 return `Desktop telemetry control '${this.operation}' failed on fd ${this.fd}.`;
134 }
135}
136
137export class DesktopTelemetryControlStalled extends Schema.TaggedError<DesktopTelemetryControlStalled>()(
138 "DesktopTelemetryControlStalled",
139 {
140 fd: Schema.Number,
141 remainingBytes: Schema.Number,
142 },
143) {
144 override get message(): string {
145 return `Desktop telemetry control stalled on fd ${this.fd} with ${this.remainingBytes} bytes remaining.`;
146 }
147}
148
149export type DesktopTelemetryControlError =
150 | DesktopTelemetryControlFailed
151 | DesktopTelemetryControlStalled;
152
153export interface DesktopTelemetryReceiverHealth {
154 readonly status: ResourceTelemetrySourceStatus;
155 readonly lastSampleAt: Option.Option<DateTime.Utc>;
156 readonly lastError: Option.Option<string>;
157}
158
159/** Moves the desktop app to `channel` and installs only `targetVersion`. */
160export interface DesktopTrackSwitch {
161 readonly channel: DesktopUpdateChannel;
162 readonly targetVersion: string;
163}
164
165export class DesktopTelemetryReceiver extends Context.Service<
166 DesktopTelemetryReceiver,
167 {
168 readonly latest: Effect.Effect<Option.Option<DesktopHostTelemetrySnapshot>>;
169 readonly changes: Stream.Stream<DesktopHostTelemetrySnapshot>;
170 readonly subscribe: Effect.Effect<
171 {
172 readonly latest: Option.Option<DesktopHostTelemetrySnapshot>;
173 readonly changes: Stream.Stream<DesktopHostTelemetrySnapshot>;
174 },
175 never,
176 Scope.Scope
177 >;
178 readonly health: Effect.Effect<DesktopTelemetryReceiverHealth>;
179 readonly subscribeHealth: Effect.Effect<
180 {
181 readonly latest: DesktopTelemetryReceiverHealth;
182 readonly changes: Stream.Stream<DesktopTelemetryReceiverHealth>;
183 },
184 never,
185 Scope.Scope
186 >;
187 readonly setDiagnosticsDemand: (
188 enabled: boolean,
189 ) => Effect.Effect<void, DesktopTelemetryControlError>;
190 readonly requestDesktopUpdate: (
191 requestId: string,
192 trackSwitch?: DesktopTrackSwitch,
193 ) => Effect.Effect<void, DesktopTelemetryControlError>;
194 readonly commitDesktopUpdate: (
195 requestId: string,
196 ) => Effect.Effect<void, DesktopTelemetryControlError>;
197 readonly cancelDesktopUpdate: (
198 requestId: string,
199 ) => Effect.Effect<void, DesktopTelemetryControlError>;
200 readonly desktopUpdates: Effect.Effect<
201 {
202 readonly latest: Option.Option<DesktopUpdateStatusReport>;
203 readonly changes: Stream.Stream<DesktopUpdateStatusReport>;
204 },
205 never,
206 Scope.Scope
207 >;
208 }
209>()("t3/resourceTelemetry/DesktopTelemetryReceiver") {}
210
211class DesktopShellCertificatesUpdateFailed extends Schema.TaggedError<DesktopShellCertificatesUpdateFailed>()(
212 "DesktopShellCertificatesUpdateFailed",
213 { cause: Schema.Defect() },
214) {
215 override get message(): string {
216 return "Failed to update desktop shell TLS certificates.";
217 }
218}
219
220export function makeApplyDesktopShellEnvironmentSnapshot(
221 environment: NodeJS.ProcessEnv,
222 platform: NodeJS.Platform,
223) {
224 let baseCertificates: Array<string> | undefined;
225 return Effect.fn("resourceTelemetry.desktopTelemetryReceiver.applyShellEnvironment")(function* (
226 message: DesktopShellEnvironmentSnapshot,
227 ) {
228 if (message.platform !== platform) return;
229 // Node's first default-CA enumeration consults process.env, so capture it before the patch.
230 if (message.environment.NODE_EXTRA_CA_CERTS !== undefined && baseCertificates === undefined) {
231 // Remove only the startup extras, preserving duplicates from bundled or system trust.
232 const extras = [...NodeTls.getCACertificates("extra")];
233 baseCertificates = NodeTls.getCACertificates("default").filter((certificate) => {
234 const index = extras.indexOf(certificate);
235 if (index === -1) return true;
236 extras.splice(index, 1);
237 return false;
238 });
239 }
240 applyShellEnvironment(environment, message.environment, platform);
241 if (message.environment.NODE_EXTRA_CA_CERTS !== undefined) {
242 yield* Effect.gen(function* () {
243 const certificates = baseCertificates;
244 // --use-openssl-ca roots are not enumerable; replacing the list would discard them.
245 if (certificates === undefined || certificates.length === 0) {
246 yield* Effect.logWarning(
247 "Preserving non-enumerable TLS trust. Restart to apply updated desktop shell certificates.",
248 );
249 return;
250 }
251 const fs = yield* FileSystem.FileSystem;
252 const file = message.environment.NODE_EXTRA_CA_CERTS;
253 const bundle = file ? yield* fs.readFileString(file) : undefined;
254 yield* Effect.try({
255 try: () => {
256 const extras =
257 bundle?.match(/-----BEGIN CERTIFICATE-----[\s\S]*?-----END CERTIFICATE-----/g) ?? [];
258 if (bundle?.trim() && extras.length === 0) {
259 throw new Error("No PEM certificates found.");
260 }
261 NodeTls.setDefaultCACertificates([...certificates, ...extras]);
262 },
263 catch: (cause) => new DesktopShellCertificatesUpdateFailed({ cause }),
264 });
265 }).pipe(
266 Effect.catch((cause) =>
267 Effect.logWarning("Failed to update desktop shell TLS certificates.", Cause.fail(cause)),
268 ),
269 );
270 }
271 if (message.discovered) delete environment.T3CODE_DESKTOP_SHELL_DISCOVERY_PENDING;
272 });
273}
274
275const decodeMessage = Schema.decodeUnknownEffect(DesktopHostTelemetryMessage);
276const encodeControlMessage = Schema.encodeEffect(
277 Schema.fromJsonString(DesktopTelemetryControlMessage),
278);
279const isDescriptorUnavailable = Schema.is(DesktopTelemetryDescriptorUnavailable);
280const isProtocolMismatch = Schema.is(DesktopTelemetryProtocolMismatch);
281const isDecodeFailed = Schema.is(DesktopTelemetryDecodeFailed);
282const isStreamFailed = Schema.is(DesktopTelemetryStreamFailed);
283
284function isDesktopTelemetryContactStale(
285 lastContactAtMs: Option.Option<number>,
286 nowMs: number,
287): boolean {
288 return Option.exists(
289 lastContactAtMs,
290 (lastContact) => nowMs - lastContact >= INITIAL_SAMPLE_DEADLINE_MS,
291 );
292}
293
294function resolveDesktopTelemetrySnapshotStaleAfterMs(
295 activeIntervalMs: number,
296 idleIntervalMs: number,
297): number {
298 return Math.max(
299 MIN_SNAPSHOT_STALE_AFTER_MS,
300 Math.max(activeIntervalMs, idleIntervalMs) + STALE_GRACE_MS,
301 );
302}
303
304function initialDesktopTelemetryContactAt(
305 desktopTelemetryFd: number | undefined,
306 nowMs: number,
307): Option.Option<number> {
308 return desktopTelemetryFd === undefined ? Option.none() : Option.some(nowMs);
309}
310
311const recordDesktopTelemetrySampleHealth = Effect.fn(
312 "resourceTelemetry.desktopTelemetryReceiver.recordSampleHealth",
313)(function* (
314 health: Ref.Ref<DesktopTelemetryReceiverHealth>,
315 healthChanges: PubSub.PubSub<DesktopTelemetryReceiverHealth>,
316 sampledAt: DateTime.Utc,
317) {
318 const next: DesktopTelemetryReceiverHealth = {
319 status: "healthy",
320 lastSampleAt: Option.some(sampledAt),
321 lastError: Option.none(),
322 };
323 yield* Ref.set(health, next);
324 yield* PubSub.publish(healthChanges, next);
325});
326
327function normalizeReceiverError(error: unknown): DesktopTelemetryReceiverError {
328 if (
329 isDescriptorUnavailable(error) ||
330 isProtocolMismatch(error) ||
331 isDecodeFailed(error) ||
332 isStreamFailed(error)
333 ) {
334 return error;
335 }
336 return new DesktopTelemetryDecodeFailed({ cause: error });
337}
338
339export const writeAllToFileDescriptor = Effect.fn(
340 "resourceTelemetry.desktopTelemetryReceiver.writeAllToFileDescriptor",
341)(function* (fd: number, payload: Buffer) {
342 let offset = 0;
343 while (offset < payload.byteLength) {
344 const written = yield* Effect.callback<number, DesktopTelemetryControlFailed>(
345 (resume, signal) => {
346 if (signal.aborted) return;
347 try {
348 NodeFS.write(
349 fd,
350 payload,
351 offset,
352 payload.byteLength - offset,
353 null,
354 (error, bytesWritten) => {
355 if (error) {
356 resume(
357 Effect.fail(
358 new DesktopTelemetryControlFailed({
359 fd,
360 operation: "write",
361 cause: error,
362 }),
363 ),
364 );
365 return;
366 }
367 resume(Effect.succeed(bytesWritten));
368 },
369 );
370 } catch (cause) {
371 resume(
372 Effect.fail(
373 new DesktopTelemetryControlFailed({
374 fd,
375 operation: "write",
376 cause,
377 }),
378 ),
379 );
380 }
381 },
382 );
383 yield* requireDesktopTelemetryWriteProgress(fd, payload.byteLength - offset, written);
384 offset += written;
385 }
386});
387
388export function requireDesktopTelemetryWriteProgress(
389 fd: number,
390 remainingBytes: number,
391 written: number,
392): Effect.Effect<void, DesktopTelemetryControlStalled> {
393 return written > 0
394 ? Effect.void
395 : Effect.fail(new DesktopTelemetryControlStalled({ fd, remainingBytes }));
396}
397
398/** @public */
399export const make = Effect.fn("resourceTelemetry.desktopTelemetryReceiver.make")(function* () {
400 const config = yield* ServerConfig.ServerConfig;
401 const environment = yield* HostProcess.Environment;
402 const platform = yield* HostProcess.Platform;
403 const applyShellSnapshot = makeApplyDesktopShellEnvironmentSnapshot(environment, platform);
404 const serverSettings = yield* ServerSettings.ServerSettingsService;
405 const latest = yield* Ref.make(Option.none<DesktopHostTelemetrySnapshot>());
406 const receiverStartedAt = yield* DateTime.now;
407 const lastContactAtMs = yield* Ref.make(
408 initialDesktopTelemetryContactAt(
409 config.desktopTelemetryFd,
410 DateTime.toEpochMillis(receiverStartedAt),
411 ),
412 );
413 const snapshotStaleAfterMs = yield* Ref.make(
414 resolveDesktopTelemetrySnapshotStaleAfterMs(
415 DEFAULT_HOST_POWER_ACTIVE_INTERVAL_MS,
416 DEFAULT_HOST_POWER_IDLE_INTERVAL_MS,
417 ),
418 );
419 const changes = yield* PubSub.sliding<DesktopHostTelemetrySnapshot>(8);
420 const healthChanges = yield* PubSub.sliding<DesktopTelemetryReceiverHealth>(4);
421 const latestUpdateReport = yield* Ref.make(Option.none<DesktopUpdateStatusReport>());
422 const updateReportChanges = yield* PubSub.sliding<DesktopUpdateStatusReport>(16);
423 const controlMutex = yield* Semaphore.make(1);
424 const snapshotMutex = yield* Semaphore.make(1);
425 const health = yield* Ref.make<DesktopTelemetryReceiverHealth>({
426 status: config.desktopTelemetryFd === undefined ? "unavailable" : "starting",
427 lastSampleAt: Option.none(),
428 lastError:
429 config.desktopTelemetryFd === undefined
430 ? Option.some(
431 new DesktopTelemetryDescriptorUnavailable({
432 mode: config.mode,
433 }).message,
434 )
435 : Option.none(),
436 });
437 const updateHealth = (
438 update: (current: DesktopTelemetryReceiverHealth) => DesktopTelemetryReceiverHealth,
439 ) =>
440 Ref.modify(health, (current) => {
441 const next = update(current);
442 return [next, next];
443 }).pipe(
444 Effect.flatMap((next) => PubSub.publish(healthChanges, next)),
445 Effect.asVoid,
446 );
447 const updateSampleHealth = (sampledAt: DateTime.Utc) =>
448 recordDesktopTelemetrySampleHealth(health, healthChanges, sampledAt);
449
450 const sendControlMessage = (message: DesktopTelemetryControlMessage) =>
451 controlMutex.withPermits(1)(
452 Effect.gen(function* () {
453 const fd = config.desktopTelemetryControlFd;
454 if (fd === undefined) return;
455 const encoded = yield* encodeControlMessage(message).pipe(
456 Effect.mapError(
457 (cause) =>
458 new DesktopTelemetryControlFailed({
459 fd,
460 operation: "encode",
461 cause,
462 }),
463 ),
464 );
465 yield* writeAllToFileDescriptor(fd, Buffer.from(`${encoded}\n`)).pipe(
466 Effect.tapError((error) =>
467 updateHealth((current) => ({
468 ...current,
469 status: "degraded",
470 lastError: Option.some(error.message),
471 })),
472 ),
473 );
474 }),
475 );
476 const setDiagnosticsDemand: DesktopTelemetryReceiver["Service"]["setDiagnosticsDemand"] = (
477 enabled,
478 ) =>
479 sendControlMessage({
480 version: 1,
481 type: "setDiagnosticsDemand",
482 enabled,
483 });
484
485 const sendHostPowerIntervals = (
486 settings: Parameters<typeof resolveServerBackgroundActivitySettings>[0],
487 ) => {
488 const resolved = resolveServerBackgroundActivitySettings(settings);
489 const activeIntervalMs = Math.max(
490 1,
491 Math.round(Duration.toMillis(resolved.hostPowerMonitorActiveInterval)),
492 );
493 const idleIntervalMs = Math.max(
494 1,
495 Math.round(Duration.toMillis(resolved.hostPowerMonitorIdleInterval)),
496 );
497 return sendControlMessage({
498 version: 1,
499 type: "setHostPowerIntervals",
500 activeIntervalMs,
501 idleIntervalMs,
502 }).pipe(
503 Effect.andThen(
504 Ref.set(
505 snapshotStaleAfterMs,
506 resolveDesktopTelemetrySnapshotStaleAfterMs(activeIntervalMs, idleIntervalMs),
507 ),
508 ),
509 );
510 };
511 if (config.desktopTelemetryControlFd !== undefined) {
512 const settingsChanges = yield* serverSettings.subscribeChanges;
513 const settings = yield* serverSettings.getSettings;
514 yield* sendHostPowerIntervals(settings).pipe(
515 Effect.catchCause((cause) =>
516 Effect.logWarning("Failed to configure desktop host-power intervals", {
517 cause: String(cause),
518 }),
519 ),
520 );
521 yield* settingsChanges.pipe(
522 Stream.runForEach((settings) =>
523 sendHostPowerIntervals(settings).pipe(
524 Effect.catchCause((cause) =>
525 Effect.logWarning("Failed to update desktop host-power intervals", {
526 cause: String(cause),
527 }),
528 ),
529 ),
530 ),
531 Effect.forkScoped,
532 );
533 }
534
535 if (config.desktopTelemetryFd !== undefined) {
536 const fd = config.desktopTelemetryFd;
537 // The parent waits before closing; use cancellable socket reads so filesystem reads cannot block exit.
538 const readable = yield* Effect.acquireRelease(
539 Effect.try({
540 try: () => new NodeNet.Socket({ fd, readable: true, writable: false }),
541 catch: (cause) => new DesktopTelemetryStreamFailed({ fd, cause }),
542 }),
543 (stream) =>
544 Effect.sync(() => {
545 stream.destroy();
546 }),
547 );
548
549 const messages: Stream.Stream<DesktopHostTelemetryMessageValue, DesktopTelemetryReceiverError> =
550 NodeStream.fromReadable<Uint8Array, DesktopTelemetryStreamFailed>({
551 evaluate: () => readable,
552 closeOnDone: true,
553 onError: (cause) => new DesktopTelemetryStreamFailed({ fd, cause }),
554 }).pipe(
555 Stream.pipeThroughChannel(Ndjson.decode({ ignoreEmptyLines: true })),
556 Stream.mapEffect(
557 (
558 value,
559 ): Effect.Effect<
560 DesktopHostTelemetryMessageValue,
561 DesktopTelemetryProtocolMismatch | DesktopTelemetryDecodeFailed
562 > => {
563 const version = messageVersion(value);
564 if (version !== undefined && version !== 1) {
565 return Effect.fail(
566 new DesktopTelemetryProtocolMismatch({
567 expectedVersion: 1,
568 receivedVersion: version,
569 }),
570 );
571 }
572 return decodeMessage(value).pipe(
573 Effect.mapError((cause) => new DesktopTelemetryDecodeFailed({ cause })),
574 );
575 },
576 ),
577 Stream.mapError(normalizeReceiverError),
578 );
579
580 yield* messages.pipe(
581 Stream.runForEach((message) => {
582 const recordContact = DateTime.now.pipe(
583 Effect.flatMap((now) =>
584 Ref.set(lastContactAtMs, Option.some(DateTime.toEpochMillis(now))),
585 ),
586 );
587 if (message.type === "desktopTelemetryHello") {
588 return recordContact.pipe(
589 Effect.andThen(
590 updateHealth((current): DesktopTelemetryReceiverHealth => ({
591 ...current,
592 status: "healthy",
593 lastError: Option.none(),
594 })),
595 ),
596 );
597 }
598
599 if (message.type === "shellEnvironment") {
600 return recordContact.pipe(Effect.andThen(applyShellSnapshot(message)));
601 }
602
603 if (message.type === "desktopUpdateStatus") {
604 return recordContact.pipe(
605 Effect.andThen(Ref.set(latestUpdateReport, Option.some(message))),
606 Effect.andThen(PubSub.publish(updateReportChanges, message)),
607 Effect.asVoid,
608 );
609 }
610
611 const sampledAt = DateTime.makeUnsafe(message.sampledAtUnixMs);
612 return snapshotMutex.withPermits(1)(
613 recordContact.pipe(
614 Effect.andThen(Ref.set(latest, Option.some(message))),
615 Effect.andThen(updateSampleHealth(sampledAt)),
616 Effect.andThen(PubSub.publish(changes, message)),
617 Effect.asVoid,
618 ),
619 );
620 }),
621 Effect.andThen(
622 updateHealth((current): DesktopTelemetryReceiverHealth => ({
623 ...current,
624 status: "stopped",
625 lastError: Option.some(new DesktopTelemetryStreamClosed({ fd }).message),
626 })),
627 ),
628 Effect.catch((error) =>
629 updateHealth((current): DesktopTelemetryReceiverHealth => ({
630 ...current,
631 status: "degraded",
632 lastError: Option.some(error.message),
633 })),
634 ),
635 Effect.forkScoped,
636 );
637
638 yield* Effect.forever(
639 Effect.sleep(STALE_CHECK_INTERVAL).pipe(
640 Effect.andThen(
641 snapshotMutex.withPermits(1)(
642 Effect.gen(function* () {
643 const now = yield* DateTime.now;
644 const nowMs = DateTime.toEpochMillis(now);
645 const staleAfterMs = yield* Ref.get(snapshotStaleAfterMs);
646 const staleSnapshot = yield* Ref.modify(latest, (current) => {
647 if (
648 Option.isNone(current) ||
649 current.value.power.stale ||
650 nowMs - current.value.sampledAtUnixMs < staleAfterMs
651 ) {
652 return [Option.none<DesktopHostTelemetrySnapshot>(), current] as const;
653 }
654 const stale: DesktopHostTelemetrySnapshot = {
655 ...current.value,
656 power: { ...current.value.power, stale: true },
657 };
658 return [Option.some(stale), Option.some(stale)] as const;
659 });
660 if (Option.isNone(staleSnapshot)) {
661 const lastContact = yield* Ref.get(lastContactAtMs);
662 if (!isDesktopTelemetryContactStale(lastContact, nowMs)) return;
663 const staleMessage = new DesktopTelemetryStale({
664 fd,
665 staleAfterMs: INITIAL_SAMPLE_DEADLINE_MS,
666 }).message;
667 const changed = yield* Ref.modify(health, (current) => {
668 if (
669 current.status === "stopped" ||
670 Option.isSome(current.lastSampleAt) ||
671 (current.status === "degraded" &&
672 Option.contains(current.lastError, staleMessage))
673 ) {
674 return [Option.none<DesktopTelemetryReceiverHealth>(), current] as const;
675 }
676 const next: DesktopTelemetryReceiverHealth = {
677 ...current,
678 status: "degraded",
679 lastError: Option.some(staleMessage),
680 };
681 return [Option.some(next), next] as const;
682 });
683 if (Option.isSome(changed)) {
684 yield* PubSub.publish(healthChanges, changed.value);
685 }
686 return;
687 }
688 yield* updateHealth((currentHealth) => ({
689 ...currentHealth,
690 status: currentHealth.status === "stopped" ? "stopped" : "degraded",
691 lastError:
692 currentHealth.status === "stopped"
693 ? currentHealth.lastError
694 : Option.some(new DesktopTelemetryStale({ fd, staleAfterMs }).message),
695 }));
696 yield* PubSub.publish(changes, staleSnapshot.value);
697 }),
698 ),
699 ),
700 ),
701 ).pipe(Effect.forkScoped);
702 }
703
704 return DesktopTelemetryReceiver.of({
705 latest: Ref.get(latest),
706 changes: Stream.fromPubSub(changes),
707 subscribe: snapshotMutex.withPermits(1)(
708 Effect.gen(function* () {
709 const initial = yield* Ref.get(latest);
710 const subscription = yield* PubSub.subscribe(changes);
711 return {
712 latest: initial,
713 changes: Stream.fromSubscription(subscription),
714 };
715 }),
716 ),
717 health: Ref.get(health),
718 subscribeHealth: subscribeBeforeSnapshotWithoutMutex(healthChanges, Ref.get(health)),
719 setDiagnosticsDemand,
720 requestDesktopUpdate: Effect.fn("DesktopTelemetryReceiver.requestDesktopUpdate")(
721 function (requestId, trackSwitch) {
722 return sendControlMessage({
723 version: 1,
724 type: "requestDesktopUpdate",
725 requestId,
726 ...trackSwitch,
727 });
728 },
729 ),
730 commitDesktopUpdate: Effect.fn("DesktopTelemetryReceiver.commitDesktopUpdate")(
731 function (requestId) {
732 return sendControlMessage({ version: 1, type: "commitDesktopUpdate", requestId });
733 },
734 ),
735 cancelDesktopUpdate: Effect.fn("DesktopTelemetryReceiver.cancelDesktopUpdate")(
736 function (requestId) {
737 return sendControlMessage({ version: 1, type: "cancelDesktopUpdate", requestId });
738 },
739 ),
740 desktopUpdates: Effect.gen(function* () {
741 const subscription = yield* PubSub.subscribe(updateReportChanges);
742 const initial = yield* Ref.get(latestUpdateReport);
743 return {
744 latest: initial,
745 changes: Stream.fromSubscription(subscription),
746 };
747 }),
748 });
749});
750
751export const layer = Layer.effect(DesktopTelemetryReceiver, make());
752
753export const layerTest = (
754 overrides: Partial<DesktopTelemetryReceiver["Service"]> = {},
755): Layer.Layer<DesktopTelemetryReceiver> => {
756 const latest = overrides.latest ?? Effect.succeedNone;
757 const changes = overrides.changes ?? Stream.empty;
758 const health =
759 overrides.health ??
760 Effect.succeed({
761 status: "unavailable" as const,
762 lastSampleAt: Option.none<DateTime.Utc>(),
763 lastError: Option.some("Desktop telemetry test implementation is unavailable."),
764 });
765 return Layer.succeed(
766 DesktopTelemetryReceiver,
767 DesktopTelemetryReceiver.of({
768 latest,
769 changes,
770 subscribe:
771 overrides.subscribe ??
772 latest.pipe(
773 Effect.map((initial) => ({
774 latest: initial,
775 changes,
776 })),
777 ),
778 health,
779 subscribeHealth:
780 overrides.subscribeHealth ??
781 health.pipe(
782 Effect.map((initial) => ({
783 latest: initial,
784 changes: Stream.empty,
785 })),
786 ),
787 setDiagnosticsDemand: Effect.fn("DesktopTelemetryReceiver.setDiagnosticsDemand")(function () {
788 return Effect.void;
789 }),
790 requestDesktopUpdate: Effect.fn("DesktopTelemetryReceiver.requestDesktopUpdate")(function () {
791 return Effect.void;
792 }),
793 commitDesktopUpdate: Effect.fn("DesktopTelemetryReceiver.commitDesktopUpdate")(function () {
794 return Effect.void;
795 }),
796 cancelDesktopUpdate: Effect.fn("DesktopTelemetryReceiver.cancelDesktopUpdate")(function () {
797 return Effect.void;
798 }),
799 desktopUpdates:
800 overrides.desktopUpdates ??
801 Effect.succeed({
802 latest: Option.none<DesktopUpdateStatusReport>(),
803 changes: Stream.empty,
804 }),
805 ...overrides,
806 }),
807 );
808};