1import {
2 DesktopHostTelemetryMessage,
3 type DesktopHostTelemetrySnapshot,
4 type DesktopTelemetryControlMessage,
5 type DesktopTelemetryCancelDesktopUpdate,
6 type DesktopTelemetryCommitDesktopUpdate,
7 type DesktopTelemetryRequestDesktopUpdate,
8 type DesktopUpdateStatusReport,
9 type HostPowerSnapshot,
10} from "@t3tools/contracts";
11import * as Cause from "effect/Cause";
12import * as Context from "effect/Context";
13import * as DateTime from "effect/DateTime";
14import * as Duration from "effect/Duration";
15import * as Effect from "effect/Effect";
16import * as Layer from "effect/Layer";
17import * as Option from "effect/Option";
18import * as PubSub from "effect/PubSub";
19import * as Queue from "effect/Queue";
20import * as Ref from "effect/Ref";
21import * as Schema from "effect/Schema";
22import * as Stream from "effect/Stream";
23
24import * as ElectronApp from "../electron/ElectronApp.ts";
25import * as ElectronPowerMonitor from "../electron/ElectronPowerMonitor.ts";
26import * as DesktopRendererHistory from "./DesktopRendererHistory.ts";
27
28const LIVE_SAMPLE_INTERVAL = Duration.seconds(1);
29const BATTERY_SAMPLE_INTERVAL = Duration.seconds(5);
30const CONSTRAINED_SAMPLE_INTERVAL = Duration.seconds(15);
31const DEFAULT_HOST_POWER_ACTIVE_INTERVAL = Duration.seconds(30);
32const DEFAULT_HOST_POWER_IDLE_INTERVAL = Duration.minutes(2);
33const IDLE_THRESHOLD_SECONDS = 60;
34const encodeMessage = Schema.encodeSync(Schema.fromJsonString(DesktopHostTelemetryMessage));
35const textEncoder = new TextEncoder();
36
37type PowerEvent =
38 | { readonly type: "locked"; readonly value: boolean }
39 | { readonly type: "suspended"; readonly value: boolean }
40 | { readonly type: "onBattery"; readonly value: boolean }
41 | { readonly type: "thermal"; readonly value: HostPowerSnapshot["thermalState"] }
42 | { readonly type: "speedLimit"; readonly value: number };
43
44interface PowerState {
45 readonly idle: HostPowerSnapshot["idle"];
46 readonly locked: HostPowerSnapshot["locked"];
47 readonly lockedEventPending: boolean;
48 readonly suspended: boolean;
49 readonly suspendedEventPending: boolean;
50 readonly onBattery: HostPowerSnapshot["onBattery"];
51 readonly onBatteryEventPending: boolean;
52 readonly thermalState: HostPowerSnapshot["thermalState"];
53 readonly speedLimitPercent: Option.Option<number>;
54}
55
56interface HostPowerIntervals {
57 readonly active: Duration.Duration;
58 readonly idle: Duration.Duration;
59}
60
61export class DesktopTelemetryPublisher extends Context.Service<
62 DesktopTelemetryPublisher,
63 {
64 readonly latest: Effect.Effect<Option.Option<DesktopHostTelemetrySnapshot>>;
65 readonly changes: Stream.Stream<DesktopHostTelemetrySnapshot>;
66 readonly encoded: Stream.Stream<Uint8Array>;
67 readonly handleControlForSource: (
68 sourceId: string,
69 message: DesktopTelemetryControlMessage,
70 ) => Effect.Effect<void>;
71 readonly removeControlSource: (sourceId: string) => Effect.Effect<void>;
72 /** Sends the report to the attached backend and replays the latest one
73 to backends that attach later (including the one spawned after a
74 relaunch). */
75 readonly publishUpdateReport: (report: DesktopUpdateStatusReport) => Effect.Effect<void>;
76 /** Update requests received over the control channel. Single consumer. */
77 readonly updateRequests: Stream.Stream<DesktopTelemetryRequestDesktopUpdate>;
78 readonly updateCommits: Stream.Stream<DesktopTelemetryCommitDesktopUpdate>;
79 readonly updateCancellations: Stream.Stream<DesktopTelemetryCancelDesktopUpdate>;
80 }
81>()("@t3tools/desktop/telemetry/DesktopTelemetryPublisher") {}
82
83function booleanState(value: boolean): HostPowerSnapshot["onBattery"] {
84 return value ? "true" : "false";
85}
86
87function idleState(value: ElectronPowerMonitor.ElectronIdleState): HostPowerSnapshot["idle"] {
88 switch (value) {
89 case "active":
90 return "false";
91 case "idle":
92 case "locked":
93 return "true";
94 case "unknown":
95 return "unknown";
96 }
97}
98
99function updatePowerState(state: PowerState, event: PowerEvent): PowerState {
100 switch (event.type) {
101 case "locked":
102 return {
103 ...state,
104 locked: booleanState(event.value),
105 lockedEventPending: true,
106 };
107 case "suspended":
108 return {
109 ...state,
110 suspended: event.value,
111 suspendedEventPending: event.value,
112 };
113 case "onBattery":
114 return {
115 ...state,
116 onBattery: booleanState(event.value),
117 onBatteryEventPending: true,
118 };
119 case "thermal":
120 return { ...state, thermalState: event.value };
121 case "speedLimit":
122 return { ...state, speedLimitPercent: Option.some(event.value) };
123 }
124}
125
126function sampleInterval(
127 power: PowerState,
128 diagnosticsDemand: boolean,
129 hostPowerIntervals: HostPowerIntervals,
130): Duration.Duration {
131 if (!diagnosticsDemand) {
132 return power.suspended || power.locked === "true" || power.idle === "true"
133 ? hostPowerIntervals.idle
134 : hostPowerIntervals.active;
135 }
136 if (
137 power.suspended ||
138 power.locked === "true" ||
139 power.thermalState === "serious" ||
140 power.thermalState === "critical"
141 ) {
142 return CONSTRAINED_SAMPLE_INTERVAL;
143 }
144 if (power.onBattery === "true") return BATTERY_SAMPLE_INTERVAL;
145 return LIVE_SAMPLE_INTERVAL;
146}
147
148/** @public Service construction is part of the canonical Effect module API. */
149export const make = Effect.fn("desktop.telemetryPublisher.make")(function* () {
150 const electronApp = yield* ElectronApp.ElectronApp;
151 const powerMonitor = yield* ElectronPowerMonitor.ElectronPowerMonitor;
152 const rendererHistory = yield* DesktopRendererHistory.DesktopRendererHistory;
153 yield* electronApp.whenReady;
154
155 const initialPowerState: PowerState = {
156 idle: "unknown",
157 locked: "unknown",
158 lockedEventPending: false,
159 suspended: false,
160 suspendedEventPending: false,
161 onBattery: booleanState(yield* powerMonitor.isOnBatteryPower),
162 onBatteryEventPending: false,
163 thermalState: yield* powerMonitor.getCurrentThermalState,
164 speedLimitPercent: Option.none(),
165 };
166 const powerState = yield* Ref.make(initialPowerState);
167 const hostPowerIntervals = yield* Ref.make<HostPowerIntervals>({
168 active: DEFAULT_HOST_POWER_ACTIVE_INTERVAL,
169 idle: DEFAULT_HOST_POWER_IDLE_INTERVAL,
170 });
171 const powerEvents = yield* Queue.unbounded<PowerEvent>();
172 const sampleTriggers = yield* Queue.sliding<void>(1);
173 const diagnosticsDemandSources = yield* Ref.make<ReadonlySet<string>>(new Set());
174 const latest = yield* Ref.make(Option.none<DesktopHostTelemetrySnapshot>());
175 const changes = yield* PubSub.sliding<DesktopHostTelemetrySnapshot>(8);
176 const sequence = yield* Ref.make(0);
177 const latestUpdateReport = yield* Ref.make(Option.none<DesktopUpdateStatusReport>());
178 const updateReportChanges = yield* PubSub.sliding<DesktopUpdateStatusReport>(16);
179 const updateRequestQueue = yield* Queue.unbounded<DesktopTelemetryRequestDesktopUpdate>();
180 const updateCommitQueue = yield* Queue.unbounded<DesktopTelemetryCommitDesktopUpdate>();
181 const updateCancellationQueue = yield* Queue.unbounded<DesktopTelemetryCancelDesktopUpdate>();
182
183 const offer = (event: PowerEvent): void => {
184 Queue.offerUnsafe(powerEvents, event);
185 };
186 yield* Effect.all(
187 [
188 powerMonitor.onSimpleEvent("lock-screen", () => offer({ type: "locked", value: true })),
189 powerMonitor.onSimpleEvent("unlock-screen", () => offer({ type: "locked", value: false })),
190 powerMonitor.onSimpleEvent("suspend", () => offer({ type: "suspended", value: true })),
191 powerMonitor.onSimpleEvent("resume", () => offer({ type: "suspended", value: false })),
192 powerMonitor.onSimpleEvent("on-battery", () => offer({ type: "onBattery", value: true })),
193 powerMonitor.onSimpleEvent("on-ac", () => offer({ type: "onBattery", value: false })),
194 powerMonitor.onThermalStateChange((value) => offer({ type: "thermal", value })),
195 powerMonitor.onSpeedLimitChange((value) => offer({ type: "speedLimit", value })),
196 ],
197 { concurrency: "unbounded" },
198 );
199 yield* Effect.forever(
200 Queue.take(powerEvents).pipe(
201 Effect.flatMap((event) => Ref.update(powerState, (state) => updatePowerState(state, event))),
202 Effect.andThen(Queue.offer(sampleTriggers, undefined)),
203 ),
204 ).pipe(Effect.forkScoped);
205
206 const sampleOnce = (allowSuspendRecovery: boolean) =>
207 Effect.gen(function* () {
208 const sampledAt = yield* DateTime.now;
209 const sampledAtUnixMs = DateTime.toEpochMillis(sampledAt);
210 const demand = (yield* Ref.get(diagnosticsDemandSources)).size > 0;
211 const [currentPower, idleSeconds, systemIdleState, onBattery, metrics] = yield* Effect.all(
212 [
213 Ref.get(powerState),
214 powerMonitor.getSystemIdleTime,
215 powerMonitor.getSystemIdleState(IDLE_THRESHOLD_SECONDS),
216 powerMonitor.isOnBatteryPower,
217 electronApp.getAppMetrics,
218 ],
219 { concurrency: "unbounded" },
220 );
221 // One Electron sampler preserves the shared CPU measurement interval.
222 // History follows the existing power-aware cadence, even without demand.
223 yield* rendererHistory.recordMetrics(metrics);
224 const polledLocked =
225 systemIdleState === "unknown"
226 ? currentPower.locked
227 : booleanState(systemIdleState === "locked");
228 const polledOnBattery = booleanState(onBattery);
229 const observedPower = yield* Ref.modify(powerState, (latestPower) => {
230 const preserveEventLocked =
231 latestPower.lockedEventPending ||
232 latestPower.locked !== currentPower.locked ||
233 latestPower.lockedEventPending !== currentPower.lockedEventPending;
234 const preserveEventOnBattery =
235 latestPower.onBatteryEventPending ||
236 latestPower.onBattery !== currentPower.onBattery ||
237 latestPower.onBatteryEventPending !== currentPower.onBatteryEventPending;
238 const preserveEventSuspended =
239 latestPower.suspendedEventPending ||
240 latestPower.suspended !== currentPower.suspended ||
241 latestPower.suspendedEventPending !== currentPower.suspendedEventPending;
242 const next: PowerState = {
243 ...latestPower,
244 idle: idleState(systemIdleState),
245 locked: preserveEventLocked ? latestPower.locked : polledLocked,
246 lockedEventPending:
247 latestPower.lockedEventPending &&
248 (systemIdleState === "unknown" || polledLocked !== latestPower.locked),
249 suspended:
250 allowSuspendRecovery && !preserveEventSuspended ? false : latestPower.suspended,
251 suspendedEventPending: false,
252 onBattery: preserveEventOnBattery ? latestPower.onBattery : polledOnBattery,
253 onBatteryEventPending:
254 latestPower.onBatteryEventPending && polledOnBattery !== latestPower.onBattery,
255 };
256 return [next, next] as const;
257 });
258 const nextSequence = yield* Ref.modify(sequence, (current) => [current + 1, current + 1]);
259 const snapshot: DesktopHostTelemetrySnapshot = {
260 version: 1,
261 type: "desktopTelemetry",
262 sequence: nextSequence,
263 sampledAtUnixMs,
264 electronPid: process.pid,
265 power: {
266 source: "electron-main",
267 idle: observedPower.idle,
268 idleSeconds,
269 locked: observedPower.locked,
270 suspended: observedPower.suspended,
271 onBattery: observedPower.onBattery,
272 lowPowerMode: "unknown",
273 thermalState: observedPower.thermalState,
274 stale: false,
275 updatedAt: sampledAt,
276 },
277 speedLimitPercent: observedPower.speedLimitPercent,
278 electronProcesses: (demand ? metrics : []).map((metric) => ({
279 pid: metric.pid,
280 creationTimeMs: Math.max(0, Math.round(metric.creationTime)),
281 type: metric.type,
282 ...(metric.name === undefined ? {} : { name: metric.name }),
283 ...(metric.serviceName === undefined ? {} : { serviceName: metric.serviceName }),
284 cpuPercent: metric.cpu.percentCPUUsage,
285 ...(metric.cpu.cumulativeCPUUsage === undefined
286 ? {}
287 : { cumulativeCpuSeconds: metric.cpu.cumulativeCPUUsage }),
288 idleWakeupsPerSecond: metric.cpu.idleWakeupsPerSecond,
289 workingSetBytes: Math.max(0, Math.round(metric.memory.workingSetSize * 1024)),
290 peakWorkingSetBytes: Math.max(0, Math.round(metric.memory.peakWorkingSetSize * 1024)),
291 })),
292 };
293
294 yield* Ref.set(latest, Option.some(snapshot));
295 yield* PubSub.publish(changes, snapshot);
296 }).pipe(
297 Effect.catchCauseIf(
298 (cause) => !Cause.hasInterrupts(cause),
299 (cause) =>
300 Effect.logWarning("Failed to sample Electron telemetry", {
301 cause: String(cause),
302 }),
303 ),
304 );
305
306 yield* Effect.gen(function* () {
307 yield* sampleOnce(false);
308 while (true) {
309 const [currentPower, demand, intervals] = yield* Effect.all([
310 Ref.get(powerState),
311 Ref.get(diagnosticsDemandSources).pipe(Effect.map((sources) => sources.size > 0)),
312 Ref.get(hostPowerIntervals),
313 ]);
314 // @effect-diagnostics-next-line raceFirstWithSleepToTimeout:off - races a trigger queue against the interval; both arms are real outcomes, not a timeout
315 const allowSuspendRecovery = yield* Effect.raceFirst(
316 Queue.take(sampleTriggers).pipe(Effect.as(false)),
317 Effect.sleep(sampleInterval(currentPower, demand, intervals)).pipe(Effect.as(true)),
318 );
319 yield* sampleOnce(allowSuspendRecovery);
320 }
321 }).pipe(Effect.forkScoped);
322
323 const handleControlForSource: DesktopTelemetryPublisher["Service"]["handleControlForSource"] = (
324 sourceId,
325 message,
326 ) => {
327 switch (message.type) {
328 case "setDiagnosticsDemand":
329 return Ref.modify(diagnosticsDemandSources, (sources) => {
330 const previous = sources.size > 0;
331 const next = new Set(sources);
332 if (message.enabled) {
333 next.add(sourceId);
334 } else {
335 next.delete(sourceId);
336 }
337 return [[previous, next.size > 0] as const, next] as const;
338 }).pipe(
339 Effect.flatMap(([previous, enabled]) =>
340 previous === enabled
341 ? Effect.void
342 : Queue.offer(sampleTriggers, undefined).pipe(Effect.asVoid),
343 ),
344 );
345 case "setHostPowerIntervals":
346 return Ref.set(hostPowerIntervals, {
347 active: Duration.millis(message.activeIntervalMs),
348 idle: Duration.millis(message.idleIntervalMs),
349 }).pipe(Effect.andThen(Queue.offer(sampleTriggers, undefined)), Effect.asVoid);
350 case "requestDesktopUpdate":
351 return Queue.offer(updateRequestQueue, message).pipe(Effect.asVoid);
352 case "commitDesktopUpdate":
353 return Queue.offer(updateCommitQueue, message).pipe(Effect.asVoid);
354 case "cancelDesktopUpdate":
355 return Queue.offer(updateCancellationQueue, message).pipe(Effect.asVoid);
356 }
357 };
358 const removeControlSource: DesktopTelemetryPublisher["Service"]["removeControlSource"] = (
359 sourceId,
360 ) =>
361 Ref.modify(diagnosticsDemandSources, (sources) => {
362 const previous = sources.size > 0;
363 const next = new Set(sources);
364 next.delete(sourceId);
365 return [[previous, next.size > 0] as const, next] as const;
366 }).pipe(
367 Effect.flatMap(([previous, enabled]) =>
368 previous === enabled
369 ? Effect.void
370 : Queue.offer(sampleTriggers, undefined).pipe(Effect.asVoid),
371 ),
372 );
373
374 const snapshots = Stream.unwrap(
375 Effect.gen(function* () {
376 const subscription = yield* PubSub.subscribe(changes);
377 const initial = yield* Ref.get(latest);
378 return Stream.concat(
379 Option.match(initial, {
380 onNone: () => Stream.empty,
381 onSome: Stream.make,
382 }),
383 Stream.fromSubscription(subscription),
384 );
385 }),
386 );
387 const updateReports = Stream.unwrap(
388 Effect.gen(function* () {
389 const subscription = yield* PubSub.subscribe(updateReportChanges);
390 const initial = yield* Ref.get(latestUpdateReport);
391 return Stream.concat(
392 Option.match(initial, {
393 onNone: () => Stream.empty,
394 onSome: Stream.make,
395 }),
396 Stream.fromSubscription(subscription),
397 );
398 }),
399 );
400 const encoded = Stream.concat(
401 Stream.make({
402 version: 1,
403 type: "desktopTelemetryHello",
404 electronPid: process.pid,
405 } as const),
406 Stream.merge(snapshots, updateReports),
407 ).pipe(Stream.map((message) => textEncoder.encode(`${encodeMessage(message)}\n`)));
408
409 const publishUpdateReport: DesktopTelemetryPublisher["Service"]["publishUpdateReport"] = (
410 report,
411 ) =>
412 Ref.set(latestUpdateReport, Option.some(report)).pipe(
413 Effect.andThen(PubSub.publish(updateReportChanges, report)),
414 Effect.asVoid,
415 );
416
417 return DesktopTelemetryPublisher.of({
418 latest: Ref.get(latest),
419 changes: Stream.fromPubSub(changes),
420 encoded,
421 handleControlForSource,
422 removeControlSource,
423 publishUpdateReport,
424 updateRequests: Stream.fromQueue(updateRequestQueue),
425 updateCommits: Stream.fromQueue(updateCommitQueue),
426 updateCancellations: Stream.fromQueue(updateCancellationQueue),
427 });
428});
429
430export const layer = Layer.effect(DesktopTelemetryPublisher, make());