1import type {
2 DesktopHostTelemetrySnapshot,
3 HostPowerSnapshot,
4 ResourceMonitorSnapshotEvent,
5 ResourceTelemetryHealth,
6 ResourceTelemetryHistoryInput,
7 ResourceTelemetryProcessIdentity,
8 ResourceTelemetryRetryResult,
9 ResourceTelemetrySnapshot,
10} from "@t3tools/contracts";
11import * as Context from "effect/Context";
12import * as DateTime from "effect/DateTime";
13import * as Effect from "effect/Effect";
14import * as Exit from "effect/Exit";
15import * as Layer from "effect/Layer";
16import * as Option from "effect/Option";
17import * as PubSub from "effect/PubSub";
18import * as Ref from "effect/Ref";
19import * as Result from "effect/Result";
20import * as Schema from "effect/Schema";
21import * as Scope from "effect/Scope";
22import * as Semaphore from "effect/Semaphore";
23import * as Stream from "effect/Stream";
24
25import * as DesktopTelemetryReceiver from "./DesktopTelemetryReceiver.ts";
26import {
27 emptyTelemetryCounters,
28 mergeProcesses,
29 type ProcessState,
30 type TelemetryCounters,
31} from "./Model.ts";
32import * as NativeTelemetryClient from "./NativeTelemetryClient.ts";
33import * as ResourceAttribution from "./ResourceAttribution.ts";
34import {
35 buildResourceTelemetryHistory,
36 normalizeResourceTelemetryHistoryInput,
37 type ResourceTelemetryHistoryWithLegacyBuckets,
38} from "./ResourceTelemetryHistory.ts";
39import { subscribeBeforeSnapshot } from "../utils/subscribeBeforeSnapshot.ts";
40
41export class ResourceTelemetryRefreshFailed extends Schema.TaggedError<ResourceTelemetryRefreshFailed>()(
42 "ResourceTelemetryRefreshFailed",
43 {
44 operation: Schema.String,
45 cause: Schema.Defect(),
46 },
47) {
48 override get message(): string {
49 return `Resource telemetry operation '${this.operation}' failed.`;
50 }
51}
52
53export class ResourceTelemetry extends Context.Service<
54 ResourceTelemetry,
55 {
56 readonly latest: Effect.Effect<ResourceTelemetrySnapshot>;
57 readonly changes: Stream.Stream<ResourceTelemetrySnapshot>;
58 readonly subscribe: Effect.Effect<
59 {
60 readonly latest: ResourceTelemetrySnapshot;
61 readonly changes: Stream.Stream<ResourceTelemetrySnapshot>;
62 },
63 never,
64 Scope.Scope
65 >;
66 readonly readHistory: (
67 input: ResourceTelemetryHistoryInput,
68 ) => Effect.Effect<ResourceTelemetryHistoryWithLegacyBuckets>;
69 readonly refresh: Effect.Effect<ResourceTelemetrySnapshot, ResourceTelemetryRefreshFailed>;
70 readonly validateProcessIdentity: (
71 identity: ResourceTelemetryProcessIdentity,
72 ) => Effect.Effect<boolean, ResourceTelemetryRefreshFailed>;
73 readonly retry: Effect.Effect<ResourceTelemetryRetryResult>;
74 }
75>()("t3/resourceTelemetry/ResourceTelemetry") {}
76
77interface TelemetryState {
78 readonly nativeSnapshot: Option.Option<ResourceMonitorSnapshotEvent>;
79 readonly desktopSnapshot: Option.Option<DesktopHostTelemetrySnapshot>;
80 readonly previous: ReadonlyMap<string, ProcessState>;
81 readonly counters: TelemetryCounters;
82 readonly latest: ResourceTelemetrySnapshot;
83 readonly lastNativeSequence: number;
84 readonly lastNativeGeneration: number;
85}
86
87interface LiveTelemetryState {
88 readonly retainCount: number;
89 readonly scope: Option.Option<Scope.Closeable>;
90}
91
92function unknownPower(updatedAt: DateTime.Utc): HostPowerSnapshot {
93 return {
94 source: "unknown",
95 idle: "unknown",
96 idleSeconds: null,
97 locked: "unknown",
98 suspended: false,
99 onBattery: "unknown",
100 lowPowerMode: "unknown",
101 thermalState: "unknown",
102 stale: true,
103 updatedAt,
104 };
105}
106
107function buildHealth(input: {
108 readonly native: NativeTelemetryClient.NativeTelemetryClientHealth;
109 readonly desktop: DesktopTelemetryReceiver.DesktopTelemetryReceiverHealth;
110 readonly nativeSnapshot: Option.Option<ResourceMonitorSnapshotEvent>;
111}): ResourceTelemetryHealth {
112 return {
113 native: {
114 status: input.native.status,
115 lastSampleAt: input.native.lastSampleAt,
116 lastError: input.native.lastError,
117 },
118 desktop: {
119 status: input.desktop.status,
120 lastSampleAt: input.desktop.lastSampleAt,
121 lastError: input.desktop.lastError,
122 },
123 sidecarVersion: Option.map(input.native.hello, (hello) => hello.sidecarVersion),
124 sidecarPid: Option.map(input.native.hello, (hello) => hello.sidecarPid),
125 restartCount: input.native.restartCount,
126 collectionDurationMicros: Option.match(input.nativeSnapshot, {
127 onNone: () => 0,
128 onSome: (snapshot) => snapshot.collectionDurationMicros,
129 }),
130 scannedProcessCount: Option.match(input.nativeSnapshot, {
131 onNone: () => 0,
132 onSome: (snapshot) => snapshot.scannedProcessCount,
133 }),
134 retainedProcessCount: Option.match(input.nativeSnapshot, {
135 onNone: () => 0,
136 onSome: (snapshot) => snapshot.retainedProcessCount,
137 }),
138 inaccessibleProcessCount: Option.match(input.nativeSnapshot, {
139 onNone: () => 0,
140 onSome: (snapshot) => snapshot.inaccessibleProcessCount,
141 }),
142 };
143}
144
145/** @public Service construction is part of the canonical Effect module API. */
146export const make = Effect.fn("resourceTelemetry.resourceTelemetry.make")(function* () {
147 const nativeClient = yield* NativeTelemetryClient.NativeTelemetryClient;
148 const desktopReceiver = yield* DesktopTelemetryReceiver.DesktopTelemetryReceiver;
149 const attribution = yield* ResourceAttribution.ResourceAttribution;
150 const mutex = yield* Semaphore.make(1);
151 const changes = yield* PubSub.sliding<ResourceTelemetrySnapshot>(8);
152 const initialReadAt = yield* DateTime.now;
153 const nativeHealthSubscription = yield* nativeClient.subscribeHealth;
154 const desktopHealthSubscription = yield* desktopReceiver.subscribeHealth;
155 const desktopSubscription = yield* desktopReceiver.subscribe;
156 const initialDesktop = desktopSubscription.latest;
157 if (Option.isSome(initialDesktop)) {
158 const electronRoot = initialDesktop.value.electronProcesses.find(
159 (process) => process.pid === initialDesktop.value.electronPid,
160 );
161 yield* nativeClient
162 .setExternalProcesses([
163 {
164 pid: initialDesktop.value.electronPid,
165 ...(electronRoot === undefined ? {} : { startTimeMs: electronRoot.creationTimeMs }),
166 },
167 ])
168 .pipe(Effect.ignore);
169 yield* nativeClient.setHostPowerState(initialDesktop.value.power).pipe(Effect.ignore);
170 }
171 const initialNativeHealth = nativeHealthSubscription.latest;
172 const initialDesktopHealth = desktopHealthSubscription.latest;
173 const initialAttribution = yield* attribution.snapshot;
174 const initialMerge = mergeProcesses({
175 serverPid: process.pid,
176 sidecarPid: Option.map(initialNativeHealth.hello, (hello) => hello.sidecarPid),
177 fallbackSampledAtMs: DateTime.toEpochMillis(initialReadAt),
178 nativeSnapshot: Option.none(),
179 desktopSnapshot: initialDesktop,
180 previous: new Map(),
181 counters: emptyTelemetryCounters(),
182 updatePrevious: false,
183 });
184 const initialSnapshot: ResourceTelemetrySnapshot = {
185 readAt: initialReadAt,
186 sampleIntervalMs: initialNativeHealth.sampleIntervalMs,
187 processes: initialMerge.processes,
188 groups: initialMerge.groups,
189 power: Option.match(initialDesktop, {
190 onNone: () => unknownPower(initialReadAt),
191 onSome: (desktop) => desktop.power,
192 }),
193 speedLimitPercent: Option.flatMap(initialDesktop, (desktop) => desktop.speedLimitPercent),
194 attribution: initialAttribution,
195 health: buildHealth({
196 native: initialNativeHealth,
197 desktop: initialDesktopHealth,
198 nativeSnapshot: Option.none(),
199 }),
200 };
201 const state = yield* Ref.make<TelemetryState>({
202 nativeSnapshot: Option.none(),
203 desktopSnapshot: initialDesktop,
204 previous: new Map(),
205 counters: emptyTelemetryCounters(),
206 latest: initialSnapshot,
207 lastNativeSequence: 0,
208 lastNativeGeneration: initialNativeHealth.restartCount,
209 });
210 const liveState = yield* Ref.make<LiveTelemetryState>({
211 retainCount: 0,
212 scope: Option.none(),
213 });
214 const liveMutex = yield* Semaphore.make(1);
215 const refreshHealth = mutex.withPermits(1)(
216 Effect.gen(function* () {
217 const current = yield* Ref.get(state);
218 const [nativeHealth, desktopHealth] = yield* Effect.all([
219 nativeClient.health,
220 desktopReceiver.health,
221 ]);
222 const snapshot: ResourceTelemetrySnapshot = {
223 ...current.latest,
224 health: buildHealth({
225 native: nativeHealth,
226 desktop: desktopHealth,
227 nativeSnapshot: current.nativeSnapshot,
228 }),
229 };
230 yield* Ref.set(state, {
231 ...current,
232 latest: snapshot,
233 });
234 if ((yield* Ref.get(liveState)).retainCount > 0) {
235 yield* PubSub.publish(changes, snapshot);
236 }
237 }),
238 );
239
240 const rebuild = (input: {
241 readonly nativeSnapshot?: NativeTelemetryClient.NativeTelemetrySnapshot;
242 readonly desktopSnapshot?: DesktopHostTelemetrySnapshot;
243 readonly updatePrevious: boolean;
244 readonly publishWhenLive?: boolean;
245 }): Effect.Effect<ResourceTelemetrySnapshot> =>
246 mutex.withPermits(1)(
247 Effect.gen(function* () {
248 const current = yield* Ref.get(state);
249 const nativeHealth = yield* nativeClient.health;
250 const incomingNativeSnapshot = input.nativeSnapshot;
251 if (
252 incomingNativeSnapshot &&
253 (incomingNativeSnapshot.generation < nativeHealth.restartCount ||
254 incomingNativeSnapshot.generation < current.lastNativeGeneration ||
255 (incomingNativeSnapshot.generation === current.lastNativeGeneration &&
256 incomingNativeSnapshot.snapshot.sequence <= current.lastNativeSequence))
257 ) {
258 return current.latest;
259 }
260 const nativeSnapshot = incomingNativeSnapshot
261 ? Option.some(incomingNativeSnapshot.snapshot)
262 : current.nativeSnapshot;
263 const desktopSnapshot = input.desktopSnapshot
264 ? Option.some(input.desktopSnapshot)
265 : current.desktopSnapshot;
266 const [desktopHealth, attributionSnapshot] = yield* Effect.all([
267 desktopReceiver.health,
268 attribution.snapshot,
269 ]);
270 const recordedElectronRoots = Option.match(nativeSnapshot, {
271 onNone: () => [],
272 onSome: (snapshot) => snapshot.externalProcesses ?? [],
273 });
274 const electronRootPids = new Set(recordedElectronRoots.map((process) => process.pid));
275 Option.match(desktopSnapshot, {
276 onNone: () => undefined,
277 onSome: (desktop) => electronRootPids.add(desktop.electronPid),
278 });
279 const electronRootStartTimes = new Map(
280 recordedElectronRoots.flatMap((process) =>
281 process.startTimeMs === undefined ? [] : [[process.pid, process.startTimeMs] as const],
282 ),
283 );
284 const merged = mergeProcesses({
285 serverPid: process.pid,
286 sidecarPid: Option.map(nativeHealth.hello, (hello) => hello.sidecarPid),
287 fallbackSampledAtMs: DateTime.toEpochMillis(current.latest.readAt),
288 nativeSnapshot,
289 desktopSnapshot,
290 electronRootPids,
291 electronRootStartTimes,
292 previous: current.previous,
293 counters: current.counters,
294 updatePrevious: input.updatePrevious,
295 });
296 const readAt = DateTime.makeUnsafe(merged.sampledAtMs);
297 const snapshot: ResourceTelemetrySnapshot = {
298 readAt,
299 sampleIntervalMs: nativeHealth.sampleIntervalMs,
300 processes: merged.processes,
301 groups: merged.groups,
302 power: Option.match(desktopSnapshot, {
303 onNone: () => unknownPower(readAt),
304 onSome: (desktop) => desktop.power,
305 }),
306 speedLimitPercent: Option.match(desktopSnapshot, {
307 onNone: () => Option.none(),
308 onSome: (desktop) => desktop.speedLimitPercent,
309 }),
310 attribution: attributionSnapshot,
311 health: buildHealth({
312 native: nativeHealth,
313 desktop: desktopHealth,
314 nativeSnapshot,
315 }),
316 };
317 yield* Ref.set(state, {
318 nativeSnapshot,
319 desktopSnapshot,
320 previous: merged.previous,
321 counters: merged.counters,
322 latest: snapshot,
323 lastNativeSequence:
324 incomingNativeSnapshot?.snapshot.sequence ?? current.lastNativeSequence,
325 lastNativeGeneration: incomingNativeSnapshot?.generation ?? current.lastNativeGeneration,
326 });
327 if (!input.publishWhenLive || (yield* Ref.get(liveState)).retainCount > 0) {
328 yield* PubSub.publish(changes, snapshot);
329 }
330 return snapshot;
331 }),
332 );
333
334 const ingestNative = (snapshot: NativeTelemetryClient.NativeTelemetrySnapshot) =>
335 rebuild({ nativeSnapshot: snapshot, updatePrevious: true });
336 const ingestDesktop = (snapshot: DesktopHostTelemetrySnapshot) =>
337 Effect.gen(function* () {
338 const electronRoot = snapshot.electronProcesses.find(
339 (process) => process.pid === snapshot.electronPid,
340 );
341 yield* nativeClient
342 .setExternalProcesses([
343 {
344 pid: snapshot.electronPid,
345 ...(electronRoot === undefined ? {} : { startTimeMs: electronRoot.creationTimeMs }),
346 },
347 ])
348 .pipe(Effect.ignore);
349 yield* nativeClient.setHostPowerState(snapshot.power).pipe(Effect.ignore);
350 return yield* rebuild({
351 desktopSnapshot: snapshot,
352 updatePrevious: false,
353 publishWhenLive: true,
354 });
355 });
356
357 yield* desktopSubscription.changes.pipe(
358 Stream.runForEach((snapshot) => ingestDesktop(snapshot)),
359 Effect.forkScoped,
360 );
361
362 const acquireLive = liveMutex.withPermits(1)(
363 Effect.uninterruptible(
364 Effect.gen(function* () {
365 const current = yield* Ref.get(liveState);
366 if (current.retainCount > 0) {
367 yield* Ref.set(liveState, { ...current, retainCount: current.retainCount + 1 });
368 return;
369 }
370
371 const scope = yield* Scope.make();
372 yield* desktopReceiver.setDiagnosticsDemand(true).pipe(Effect.ignore);
373 yield* nativeClient.snapshots.pipe(
374 Stream.runForEach(ingestNative),
375 Effect.catch((error) =>
376 Effect.logWarning("Native resource telemetry stream stopped", {
377 cause: error.message,
378 }),
379 ),
380 Effect.forkIn(scope),
381 );
382 yield* nativeClient.sampleNow.pipe(
383 Effect.flatMap(ingestNative),
384 Effect.ignore,
385 Effect.forkIn(scope),
386 );
387 yield* Ref.set(liveState, { retainCount: 1, scope: Option.some(scope) });
388 }),
389 ),
390 );
391
392 const releaseLive = liveMutex.withPermits(1)(
393 Effect.gen(function* () {
394 const current = yield* Ref.get(liveState);
395 if (current.retainCount <= 1) {
396 yield* Ref.set(liveState, { retainCount: 0, scope: Option.none() });
397 if (Option.isSome(current.scope)) {
398 yield* Scope.close(current.scope.value, Exit.void).pipe(Effect.ignore);
399 }
400 yield* desktopReceiver.setDiagnosticsDemand(false).pipe(Effect.ignore);
401 return;
402 }
403 yield* Ref.set(liveState, { ...current, retainCount: current.retainCount - 1 });
404 }),
405 );
406
407 const latest = Ref.get(state).pipe(Effect.map((current) => current.latest));
408 const subscribe = subscribeBeforeSnapshot(
409 changes,
410 Effect.acquireRelease(acquireLive, () => releaseLive).pipe(Effect.andThen(latest)),
411 mutex,
412 );
413 const liveChanges = Stream.unwrap(Effect.map(subscribe, ({ changes }) => changes));
414
415 const readHistory: ResourceTelemetry["Service"]["readHistory"] = (input) =>
416 Effect.gen(function* () {
417 const readAt = yield* DateTime.now;
418 const normalizedInput = normalizeResourceTelemetryHistoryInput(input);
419 const historyResult = yield* Effect.result(
420 nativeClient.readHistory(normalizedInput.windowMs),
421 );
422 if (Result.isFailure(historyResult)) {
423 yield* Effect.logWarning("Failed to read native resource telemetry history", {
424 cause: historyResult.failure.message,
425 });
426 }
427 const [nativeHealth, desktopHealth] = yield* Effect.all([
428 nativeClient.health,
429 desktopReceiver.health,
430 ]);
431 const current = yield* Ref.get(state);
432 return buildResourceTelemetryHistory({
433 readAt,
434 windowMs: normalizedInput.windowMs,
435 bucketMs: normalizedInput.bucketMs,
436 sampleIntervalMs: nativeHealth.sampleIntervalMs,
437 serverPid: process.pid,
438 sidecarPid: Option.map(nativeHealth.hello, (hello) => hello.sidecarPid),
439 desktopSnapshot: current.desktopSnapshot,
440 snapshots: Result.isSuccess(historyResult) ? historyResult.success : [],
441 health: buildHealth({
442 native: nativeHealth,
443 desktop: desktopHealth,
444 nativeSnapshot: current.nativeSnapshot,
445 }),
446 });
447 });
448 yield* nativeHealthSubscription.changes.pipe(
449 Stream.runForEach(() => refreshHealth),
450 Effect.forkScoped,
451 );
452 yield* desktopHealthSubscription.changes.pipe(
453 Stream.runForEach(() => refreshHealth),
454 Effect.forkScoped,
455 );
456
457 const refresh: ResourceTelemetry["Service"]["refresh"] = nativeClient.sampleNow.pipe(
458 Effect.flatMap(ingestNative),
459 Effect.mapError(
460 (cause) =>
461 new ResourceTelemetryRefreshFailed({
462 operation: "refresh",
463 cause,
464 }),
465 ),
466 );
467
468 const validateProcessIdentity: ResourceTelemetry["Service"]["validateProcessIdentity"] = (
469 identity,
470 ) =>
471 nativeClient.sampleNow.pipe(
472 Effect.map(({ snapshot }) =>
473 snapshot.processes.some(
474 (process) => process.pid === identity.pid && process.startTimeMs === identity.startTimeMs,
475 ),
476 ),
477 Effect.mapError(
478 (cause) =>
479 new ResourceTelemetryRefreshFailed({
480 operation: "validateProcessIdentity",
481 cause,
482 }),
483 ),
484 );
485
486 return ResourceTelemetry.of({
487 latest,
488 changes: liveChanges,
489 subscribe,
490 readHistory,
491 refresh,
492 validateProcessIdentity,
493 retry: nativeClient.retry.pipe(
494 Effect.zip(Ref.get(state)),
495 Effect.map(([accepted, current]): ResourceTelemetryRetryResult => ({
496 accepted,
497 snapshot: current.latest,
498 })),
499 ),
500 });
501});
502
503export const layer = Layer.effect(ResourceTelemetry, make());