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