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