1import type {
2 DesktopHostTelemetrySnapshot,
3 ResourceMonitorSnapshotEvent,
4 ResourceTelemetryHealth,
5 ResourceTelemetryHistory,
6 ResourceTelemetryHistoryBucket,
7 ResourceTelemetryProcess,
8 ResourceTelemetryProcessSummary,
9} from "@t3tools/contracts";
10import * as DateTime from "effect/DateTime";
11import * as Option from "effect/Option";
12
13import {
14 emptyTelemetryCounters,
15 mergeProcesses,
16 processIdentityKey,
17 type ProcessState,
18} from "./Model.ts";
19
20const MAX_HISTORY_WINDOW_MS = 60 * 60_000;
21
22export function normalizeResourceTelemetryHistoryInput(input: {
23 readonly windowMs: number;
24 readonly bucketMs: number;
25}): { readonly windowMs: number; readonly bucketMs: number } {
26 const windowMs = Math.max(1_000, Math.min(MAX_HISTORY_WINDOW_MS, input.windowMs));
27 return {
28 windowMs,
29 bucketMs: Math.max(1_000, Math.min(windowMs, input.bucketMs)),
30 };
31}
32
33interface AggregateSample {
34 readonly sampledAtMs: number;
35 readonly cpuPercent: number;
36 readonly rssBytes: number;
37 readonly processCount: number;
38 readonly ioReadBytes: number;
39 readonly ioWriteBytes: number;
40}
41
42interface ProcessSample {
43 readonly sampledAtMs: number;
44 readonly process: ResourceTelemetryProcess;
45 readonly cpuTimeMs: number;
46 readonly ioReadBytes: number;
47 readonly ioWriteBytes: number;
48}
49
50export interface BuildResourceTelemetryHistoryInput {
51 readonly readAt: DateTime.Utc;
52 readonly windowMs: number;
53 readonly bucketMs: number;
54 readonly sampleIntervalMs: number;
55 readonly serverPid: number;
56 readonly sidecarPid: Option.Option<number>;
57 readonly desktopSnapshot: Option.Option<DesktopHostTelemetrySnapshot>;
58 readonly snapshots: ReadonlyArray<ResourceMonitorSnapshotEvent>;
59 readonly health: ResourceTelemetryHealth;
60}
61
62export type ResourceTelemetryHistoryWithLegacyBuckets = ResourceTelemetryHistory & {
63 readonly legacyBackendBuckets?: ReadonlyArray<ResourceTelemetryHistoryBucket>;
64};
65
66function summarizeProcesses(
67 samples: ReadonlyArray<ProcessSample>,
68): ReadonlyArray<ResourceTelemetryProcessSummary> {
69 const groups = new Map<string, ProcessSample[]>();
70 for (const sample of samples) {
71 const identityKey = processIdentityKey(
72 sample.process.identity.pid,
73 sample.process.identity.startTimeMs,
74 );
75 const current = groups.get(identityKey) ?? [];
76 current.push(sample);
77 groups.set(identityKey, current);
78 }
79
80 return [...groups.values()]
81 .map((processSamples): ResourceTelemetryProcessSummary => {
82 const sorted = processSamples.toSorted((left, right) => left.sampledAtMs - right.sampledAtMs);
83 const first = sorted[0]!;
84 const latest = sorted[sorted.length - 1]!;
85 const cpuTotal = sorted.reduce((total, sample) => total + sample.process.cpuPercent, 0);
86 return {
87 identity: latest.process.identity,
88 ppid: latest.process.ppid,
89 depth: latest.process.depth,
90 name: latest.process.name,
91 command: latest.process.command,
92 category: latest.process.category,
93 firstSeenAt: first.process.firstSeenAt,
94 lastSeenAt: latest.process.lastSeenAt,
95 currentCpuPercent: latest.process.cpuPercent,
96 avgCpuPercent: cpuTotal / sorted.length,
97 maxCpuPercent: Math.max(...sorted.map((sample) => sample.process.cpuPercent)),
98 cpuTimeMs: sorted.reduce((total, sample) => total + sample.cpuTimeMs, 0),
99 currentRssBytes: latest.process.residentBytes,
100 peakRssBytes: Math.max(...sorted.map((sample) => sample.process.residentBytes)),
101 ioReadBytes: sorted.reduce((total, sample) => total + sample.ioReadBytes, 0),
102 ioWriteBytes: sorted.reduce((total, sample) => total + sample.ioWriteBytes, 0),
103 ioSemantics: latest.process.ioSemantics,
104 sampleCount: sorted.length,
105 };
106 })
107 .toSorted(
108 (left, right) => right.cpuTimeMs - left.cpuTimeMs || right.peakRssBytes - left.peakRssBytes,
109 );
110}
111
112function buildBuckets(input: {
113 readonly samples: ReadonlyArray<AggregateSample>;
114 readonly nowMs: number;
115 readonly windowMs: number;
116 readonly bucketMs: number;
117}): ReadonlyArray<ResourceTelemetryHistoryBucket> {
118 const windowStartMs = input.nowMs - input.windowMs;
119 const buckets: ResourceTelemetryHistoryBucket[] = [];
120 for (let startedAtMs = windowStartMs; startedAtMs < input.nowMs; startedAtMs += input.bucketMs) {
121 const endedAtMs = Math.min(input.nowMs, startedAtMs + input.bucketMs);
122 const samples = input.samples.filter(
123 (sample) =>
124 sample.sampledAtMs >= startedAtMs &&
125 (endedAtMs === input.nowMs
126 ? sample.sampledAtMs <= endedAtMs
127 : sample.sampledAtMs < endedAtMs),
128 );
129 const cpuTotal = samples.reduce((total, sample) => total + sample.cpuPercent, 0);
130 buckets.push({
131 startedAt: DateTime.makeUnsafe(startedAtMs),
132 endedAt: DateTime.makeUnsafe(endedAtMs),
133 avgCpuPercent: samples.length === 0 ? 0 : cpuTotal / samples.length,
134 maxCpuPercent:
135 samples.length === 0 ? 0 : Math.max(...samples.map((sample) => sample.cpuPercent)),
136 maxRssBytes: samples.length === 0 ? 0 : Math.max(...samples.map((sample) => sample.rssBytes)),
137 ioReadBytes: samples.reduce((total, sample) => total + sample.ioReadBytes, 0),
138 ioWriteBytes: samples.reduce((total, sample) => total + sample.ioWriteBytes, 0),
139 maxProcessCount:
140 samples.length === 0 ? 0 : Math.max(...samples.map((sample) => sample.processCount)),
141 });
142 }
143 return buckets;
144}
145
146export function buildResourceTelemetryHistory(
147 input: BuildResourceTelemetryHistoryInput,
148): ResourceTelemetryHistoryWithLegacyBuckets & {
149 readonly legacyBackendBuckets: ReadonlyArray<ResourceTelemetryHistoryBucket>;
150} {
151 const readAtMs = DateTime.toEpochMillis(input.readAt);
152 const { windowMs, bucketMs } = normalizeResourceTelemetryHistoryInput(input);
153 const windowStartMs = readAtMs - windowMs;
154 const eligibleSnapshots = input.snapshots
155 .filter((snapshot) => snapshot.sampledAtUnixMs <= readAtMs)
156 .toSorted((left, right) => left.sampledAtUnixMs - right.sampledAtUnixMs);
157 const snapshotsInWindow = eligibleSnapshots.filter(
158 (snapshot) => snapshot.sampledAtUnixMs >= windowStartMs,
159 );
160 const precedingSnapshot = eligibleSnapshots.findLast(
161 (snapshot) => snapshot.sampledAtUnixMs < windowStartMs,
162 );
163 const snapshots = precedingSnapshot
164 ? [precedingSnapshot, ...snapshotsInWindow]
165 : snapshotsInWindow;
166 const aggregateSamples: AggregateSample[] = [];
167 const legacyBackendAggregateSamples: AggregateSample[] = [];
168 const processSamples: ProcessSample[] = [];
169 const previous = new Map<string, ProcessState>();
170 const counters = emptyTelemetryCounters();
171 let previousSnapshotAtMs: number | undefined;
172
173 for (const snapshot of snapshots) {
174 const deltaWindowFraction =
175 previousSnapshotAtMs !== undefined &&
176 previousSnapshotAtMs < windowStartMs &&
177 snapshot.sampledAtUnixMs > previousSnapshotAtMs
178 ? Math.max(
179 0,
180 Math.min(
181 1,
182 (snapshot.sampledAtUnixMs - windowStartMs) /
183 (snapshot.sampledAtUnixMs - previousSnapshotAtMs),
184 ),
185 )
186 : 1;
187 previousSnapshotAtMs = snapshot.sampledAtUnixMs;
188 const recordedExternalProcesses =
189 snapshot.externalProcesses ??
190 Option.match(input.desktopSnapshot, {
191 onNone: () => [],
192 onSome: (desktopSnapshot) => [
193 {
194 pid: desktopSnapshot.electronPid,
195 startTimeMs: desktopSnapshot.electronProcesses.find(
196 (metric) => metric.pid === desktopSnapshot.electronPid,
197 )?.creationTimeMs,
198 },
199 ],
200 });
201 const electronRootPids = new Set(recordedExternalProcesses.map((process) => process.pid));
202 const electronRootStartTimes = new Map(
203 recordedExternalProcesses.flatMap((process) =>
204 process.startTimeMs === undefined ? [] : [[process.pid, process.startTimeMs] as const],
205 ),
206 );
207 const merged = mergeProcesses({
208 serverPid: input.serverPid,
209 sidecarPid: input.sidecarPid,
210 fallbackSampledAtMs: snapshot.sampledAtUnixMs,
211 nativeSnapshot: Option.some(snapshot),
212 desktopSnapshot: Option.none(),
213 electronRootPids,
214 electronRootStartTimes,
215 previous,
216 counters,
217 updatePrevious: true,
218 accumulateCounters: false,
219 });
220 // Keep absent identities' baselines without copying every exited process on each sample.
221 for (const [identityKey, state] of merged.previous) {
222 previous.set(identityKey, state);
223 }
224 if (snapshot.sampledAtUnixMs < windowStartMs) {
225 continue;
226 }
227 const deltas =
228 deltaWindowFraction === 1
229 ? merged.deltas
230 : merged.deltas.map((delta) => ({
231 ...delta,
232 cpuTimeMs: Math.round(delta.cpuTimeMs * deltaWindowFraction),
233 ioReadBytes: Math.round(delta.ioReadBytes * deltaWindowFraction),
234 ioWriteBytes: Math.round(delta.ioWriteBytes * deltaWindowFraction),
235 }));
236 const deltasByIdentity = new Map(
237 deltas.map((processDelta) => [processDelta.identityKey, processDelta]),
238 );
239 aggregateSamples.push({
240 sampledAtMs: snapshot.sampledAtUnixMs,
241 cpuPercent: merged.groups.allT3.currentCpuPercent,
242 rssBytes: merged.groups.allT3.currentRssBytes,
243 processCount: merged.groups.allT3.processCount,
244 ioReadBytes: deltas.reduce((total, process) => total + process.ioReadBytes, 0),
245 ioWriteBytes: deltas.reduce((total, process) => total + process.ioWriteBytes, 0),
246 });
247 const backendDeltas = deltas.filter(
248 (processDelta) =>
249 processDelta.category === "server" ||
250 processDelta.category === "server-child" ||
251 processDelta.category === "provider-root" ||
252 processDelta.category === "terminal-root",
253 );
254 legacyBackendAggregateSamples.push({
255 sampledAtMs: snapshot.sampledAtUnixMs,
256 cpuPercent: merged.groups.backend.currentCpuPercent,
257 rssBytes: merged.groups.backend.currentRssBytes,
258 processCount: merged.groups.backend.processCount,
259 ioReadBytes: backendDeltas.reduce((total, process) => total + process.ioReadBytes, 0),
260 ioWriteBytes: backendDeltas.reduce((total, process) => total + process.ioWriteBytes, 0),
261 });
262 for (const process of merged.processes) {
263 const processDelta = deltasByIdentity.get(
264 processIdentityKey(process.identity.pid, process.identity.startTimeMs),
265 );
266 processSamples.push({
267 sampledAtMs: snapshot.sampledAtUnixMs,
268 process,
269 cpuTimeMs: processDelta?.cpuTimeMs ?? 0,
270 ioReadBytes: processDelta?.ioReadBytes ?? 0,
271 ioWriteBytes: processDelta?.ioWriteBytes ?? 0,
272 });
273 }
274 }
275
276 return {
277 readAt: input.readAt,
278 windowMs,
279 bucketMs,
280 sampleIntervalMs: input.sampleIntervalMs,
281 retainedSampleCount: aggregateSamples.length + processSamples.length,
282 buckets: buildBuckets({ samples: aggregateSamples, nowMs: readAtMs, windowMs, bucketMs }),
283 legacyBackendBuckets: buildBuckets({
284 samples: legacyBackendAggregateSamples,
285 nowMs: readAtMs,
286 windowMs,
287 bucketMs,
288 }),
289 topProcesses: summarizeProcesses(processSamples),
290 health: input.health,
291 };
292}