1import type {
2 ResourceTelemetryProcessCategory,
3 ServerProcessResourceHistoryInput,
4 ServerProcessResourceHistoryResult,
5} from "@t3tools/contracts";
6import * as Context from "effect/Context";
7import * as Effect from "effect/Effect";
8import * as Layer from "effect/Layer";
9import * as Option from "effect/Option";
10
11import * as ResourceTelemetry from "../resourceTelemetry/ResourceTelemetry.ts";
12
13export class ProcessResourceMonitor extends Context.Service<
14 ProcessResourceMonitor,
15 {
16 readonly readHistory: (
17 input: ServerProcessResourceHistoryInput,
18 ) => Effect.Effect<ServerProcessResourceHistoryResult>;
19 }
20>()("t3/diagnostics/ProcessResourceMonitor") {}
21
22function isLegacyBackendCategory(category: ResourceTelemetryProcessCategory): boolean {
23 return (
24 category === "server" ||
25 category === "server-child" ||
26 category === "provider-root" ||
27 category === "terminal-root"
28 );
29}
30
31/** @public Service construction is part of the canonical Effect module API. */
32export const make = Effect.fn("makeProcessResourceMonitor")(function* () {
33 const telemetry = yield* ResourceTelemetry.ResourceTelemetry;
34 const readHistory: ProcessResourceMonitor["Service"]["readHistory"] = (input) =>
35 telemetry.readHistory(input).pipe(
36 Effect.map((history) => {
37 const topProcesses = history.topProcesses
38 .filter((entry) => isLegacyBackendCategory(entry.category))
39 .map((entry) => ({
40 processKey: `${entry.identity.pid}:${entry.identity.startTimeMs}`,
41 pid: entry.identity.pid,
42 ppid: entry.ppid,
43 command: entry.command || entry.name || "unknown",
44 depth: entry.depth,
45 isServerRoot: entry.category === "server",
46 firstSeenAt: entry.firstSeenAt,
47 lastSeenAt: entry.lastSeenAt,
48 currentCpuPercent: entry.currentCpuPercent,
49 avgCpuPercent: entry.avgCpuPercent,
50 maxCpuPercent: entry.maxCpuPercent,
51 cpuSecondsApprox: entry.cpuTimeMs / 1_000,
52 currentRssBytes: entry.currentRssBytes,
53 maxRssBytes: entry.peakRssBytes,
54 sampleCount: entry.sampleCount,
55 }));
56 return {
57 readAt: history.readAt,
58 windowMs: history.windowMs,
59 bucketMs: history.bucketMs,
60 sampleIntervalMs: history.sampleIntervalMs,
61 retainedSampleCount: history.retainedSampleCount,
62 totalCpuSecondsApprox: topProcesses.reduce(
63 (total, entry) => total + entry.cpuSecondsApprox,
64 0,
65 ),
66 buckets: (history.legacyBackendBuckets ?? history.buckets).map((bucket) => ({
67 startedAt: bucket.startedAt,
68 endedAt: bucket.endedAt,
69 avgCpuPercent: bucket.avgCpuPercent,
70 maxCpuPercent: bucket.maxCpuPercent,
71 maxRssBytes: bucket.maxRssBytes,
72 maxProcessCount: bucket.maxProcessCount,
73 })),
74 topProcesses,
75 error: history.health.native.lastError.pipe(
76 Option.map((message) => ({
77 failureTag: "ProcessDiagnosticsQueryFailedError" as const,
78 message,
79 })),
80 ),
81 };
82 }),
83 );
84
85 return ProcessResourceMonitor.of({ readHistory });
86});
87
88export const layer = Layer.effect(ProcessResourceMonitor, make());