import type {  DesktopHostTelemetrySnapshot,  HostPowerSnapshot,  ResourceMonitorSnapshotEvent,  ResourceTelemetryHealth,  ResourceTelemetryHistoryInput,  ResourceTelemetryProcessIdentity,  ResourceTelemetryRetryResult,  ResourceTelemetrySnapshot,} from "@t3tools/contracts";import * as Context from "effect/Context";import * as DateTime from "effect/DateTime";import * as Effect from "effect/Effect";import * as Exit from "effect/Exit";import * as Layer from "effect/Layer";import * as Option from "effect/Option";import * as PubSub from "effect/PubSub";import * as Ref from "effect/Ref";import * as Result from "effect/Result";import * as Schema from "effect/Schema";import * as Scope from "effect/Scope";import * as Semaphore from "effect/Semaphore";import * as Stream from "effect/Stream"; import * as DesktopTelemetryReceiver from "./DesktopTelemetryReceiver.ts";import {  emptyTelemetryCounters,  mergeProcesses,  type ProcessState,  type TelemetryCounters,} from "./Model.ts";import * as NativeTelemetryClient from "./NativeTelemetryClient.ts";import * as ResourceAttribution from "./ResourceAttribution.ts";import {  buildResourceTelemetryHistory,  normalizeResourceTelemetryHistoryInput,  type ResourceTelemetryHistoryWithLegacyBuckets,} from "./ResourceTelemetryHistory.ts";import { subscribeBeforeSnapshot } from "../utils/subscribeBeforeSnapshot.ts"; export class ResourceTelemetryRefreshFailed extends Schema.TaggedErrorClass<ResourceTelemetryRefreshFailed>()(  "ResourceTelemetryRefreshFailed",  {    operation: Schema.String,    cause: Schema.Defect(),  },) {  override get message(): string {    return `Resource telemetry operation '${this.operation}' failed.`;  }} export class ResourceTelemetry extends Context.Service<  ResourceTelemetry,  {    readonly latest: Effect.Effect<ResourceTelemetrySnapshot>;    readonly changes: Stream.Stream<ResourceTelemetrySnapshot>;    readonly subscribe: Effect.Effect<      {        readonly latest: ResourceTelemetrySnapshot;        readonly changes: Stream.Stream<ResourceTelemetrySnapshot>;      },      never,      Scope.Scope    >;    readonly readHistory: (      input: ResourceTelemetryHistoryInput,    ) => Effect.Effect<ResourceTelemetryHistoryWithLegacyBuckets>;    readonly refresh: Effect.Effect<ResourceTelemetrySnapshot, ResourceTelemetryRefreshFailed>;    readonly validateProcessIdentity: (      identity: ResourceTelemetryProcessIdentity,    ) => Effect.Effect<boolean, ResourceTelemetryRefreshFailed>;    readonly retry: Effect.Effect<ResourceTelemetryRetryResult>;  }>()("t3/resourceTelemetry/ResourceTelemetry") {} interface TelemetryState {  readonly nativeSnapshot: Option.Option<ResourceMonitorSnapshotEvent>;  readonly desktopSnapshot: Option.Option<DesktopHostTelemetrySnapshot>;  readonly previous: ReadonlyMap<string, ProcessState>;  readonly counters: TelemetryCounters;  readonly latest: ResourceTelemetrySnapshot;  readonly lastNativeSequence: number;  readonly lastNativeGeneration: number;} interface LiveTelemetryState {  readonly retainCount: number;  readonly scope: Option.Option<Scope.Closeable>;} function unknownPower(updatedAt: DateTime.Utc): HostPowerSnapshot {  return {    source: "unknown",    idle: "unknown",    idleSeconds: null,    locked: "unknown",    suspended: false,    onBattery: "unknown",    lowPowerMode: "unknown",    thermalState: "unknown",    stale: true,    updatedAt,  };} function buildHealth(input: {  readonly native: NativeTelemetryClient.NativeTelemetryClientHealth;  readonly desktop: DesktopTelemetryReceiver.DesktopTelemetryReceiverHealth;  readonly nativeSnapshot: Option.Option<ResourceMonitorSnapshotEvent>;}): ResourceTelemetryHealth {  return {    native: {      status: input.native.status,      lastSampleAt: input.native.lastSampleAt,      lastError: input.native.lastError,    },    desktop: {      status: input.desktop.status,      lastSampleAt: input.desktop.lastSampleAt,      lastError: input.desktop.lastError,    },    sidecarVersion: Option.map(input.native.hello, (hello) => hello.sidecarVersion),    sidecarPid: Option.map(input.native.hello, (hello) => hello.sidecarPid),    restartCount: input.native.restartCount,    collectionDurationMicros: Option.match(input.nativeSnapshot, {      onNone: () => 0,      onSome: (snapshot) => snapshot.collectionDurationMicros,    }),    scannedProcessCount: Option.match(input.nativeSnapshot, {      onNone: () => 0,      onSome: (snapshot) => snapshot.scannedProcessCount,    }),    retainedProcessCount: Option.match(input.nativeSnapshot, {      onNone: () => 0,      onSome: (snapshot) => snapshot.retainedProcessCount,    }),    inaccessibleProcessCount: Option.match(input.nativeSnapshot, {      onNone: () => 0,      onSome: (snapshot) => snapshot.inaccessibleProcessCount,    }),  };} export const make = Effect.fn("resourceTelemetry.resourceTelemetry.make")(function* () {  const nativeClient = yield* NativeTelemetryClient.NativeTelemetryClient;  const desktopReceiver = yield* DesktopTelemetryReceiver.DesktopTelemetryReceiver;  const attribution = yield* ResourceAttribution.ResourceAttribution;  const mutex = yield* Semaphore.make(1);  const changes = yield* PubSub.sliding<ResourceTelemetrySnapshot>(8);  const initialReadAt = yield* DateTime.now;  const nativeHealthSubscription = yield* nativeClient.subscribeHealth;  const desktopHealthSubscription = yield* desktopReceiver.subscribeHealth;  const desktopSubscription = yield* desktopReceiver.subscribe;  const initialDesktop = desktopSubscription.latest;  if (Option.isSome(initialDesktop)) {    const electronRoot = initialDesktop.value.electronProcesses.find(      (process) => process.pid === initialDesktop.value.electronPid,    );    yield* nativeClient      .setExternalProcesses([        {          pid: initialDesktop.value.electronPid,          ...(electronRoot === undefined ? {} : { startTimeMs: electronRoot.creationTimeMs }),        },      ])      .pipe(Effect.ignore);    yield* nativeClient.setHostPowerState(initialDesktop.value.power).pipe(Effect.ignore);  }  const initialNativeHealth = nativeHealthSubscription.latest;  const initialDesktopHealth = desktopHealthSubscription.latest;  const initialAttribution = yield* attribution.snapshot;  const initialMerge = mergeProcesses({    serverPid: process.pid,    sidecarPid: Option.map(initialNativeHealth.hello, (hello) => hello.sidecarPid),    fallbackSampledAtMs: DateTime.toEpochMillis(initialReadAt),    nativeSnapshot: Option.none(),    desktopSnapshot: initialDesktop,    previous: new Map(),    counters: emptyTelemetryCounters(),    updatePrevious: false,  });  const initialSnapshot: ResourceTelemetrySnapshot = {    readAt: initialReadAt,    sampleIntervalMs: initialNativeHealth.sampleIntervalMs,    processes: initialMerge.processes,    groups: initialMerge.groups,    power: Option.match(initialDesktop, {      onNone: () => unknownPower(initialReadAt),      onSome: (desktop) => desktop.power,    }),    speedLimitPercent: Option.flatMap(initialDesktop, (desktop) => desktop.speedLimitPercent),    attribution: initialAttribution,    health: buildHealth({      native: initialNativeHealth,      desktop: initialDesktopHealth,      nativeSnapshot: Option.none(),    }),  };  const state = yield* Ref.make<TelemetryState>({    nativeSnapshot: Option.none(),    desktopSnapshot: initialDesktop,    previous: new Map(),    counters: emptyTelemetryCounters(),    latest: initialSnapshot,    lastNativeSequence: 0,    lastNativeGeneration: initialNativeHealth.restartCount,  });  const liveState = yield* Ref.make<LiveTelemetryState>({    retainCount: 0,    scope: Option.none(),  });  const liveMutex = yield* Semaphore.make(1);  const refreshHealth = mutex.withPermits(1)(    Effect.gen(function* () {      const current = yield* Ref.get(state);      const [nativeHealth, desktopHealth] = yield* Effect.all([        nativeClient.health,        desktopReceiver.health,      ]);      const snapshot: ResourceTelemetrySnapshot = {        ...current.latest,        health: buildHealth({          native: nativeHealth,          desktop: desktopHealth,          nativeSnapshot: current.nativeSnapshot,        }),      };      yield* Ref.set(state, {        ...current,        latest: snapshot,      });      if ((yield* Ref.get(liveState)).retainCount > 0) {        yield* PubSub.publish(changes, snapshot);      }    }),  );   const rebuild = (input: {    readonly nativeSnapshot?: NativeTelemetryClient.NativeTelemetrySnapshot;    readonly desktopSnapshot?: DesktopHostTelemetrySnapshot;    readonly updatePrevious: boolean;    readonly publishWhenLive?: boolean;  }): Effect.Effect<ResourceTelemetrySnapshot> =>    mutex.withPermits(1)(      Effect.gen(function* () {        const current = yield* Ref.get(state);        const nativeHealth = yield* nativeClient.health;        const incomingNativeSnapshot = input.nativeSnapshot;        if (          incomingNativeSnapshot &&          (incomingNativeSnapshot.generation < nativeHealth.restartCount ||            incomingNativeSnapshot.generation < current.lastNativeGeneration ||            (incomingNativeSnapshot.generation === current.lastNativeGeneration &&              incomingNativeSnapshot.snapshot.sequence <= current.lastNativeSequence))        ) {          return current.latest;        }        const nativeSnapshot = incomingNativeSnapshot          ? Option.some(incomingNativeSnapshot.snapshot)          : current.nativeSnapshot;        const desktopSnapshot = input.desktopSnapshot          ? Option.some(input.desktopSnapshot)          : current.desktopSnapshot;        const [desktopHealth, attributionSnapshot] = yield* Effect.all([          desktopReceiver.health,          attribution.snapshot,        ]);        const recordedElectronRoots = Option.match(nativeSnapshot, {          onNone: () => [],          onSome: (snapshot) => snapshot.externalProcesses ?? [],        });        const electronRootPids = new Set(recordedElectronRoots.map((process) => process.pid));        Option.match(desktopSnapshot, {          onNone: () => undefined,          onSome: (desktop) => electronRootPids.add(desktop.electronPid),        });        const electronRootStartTimes = new Map(          recordedElectronRoots.flatMap((process) =>            process.startTimeMs === undefined ? [] : [[process.pid, process.startTimeMs] as const],          ),        );        const merged = mergeProcesses({          serverPid: process.pid,          sidecarPid: Option.map(nativeHealth.hello, (hello) => hello.sidecarPid),          fallbackSampledAtMs: DateTime.toEpochMillis(current.latest.readAt),          nativeSnapshot,          desktopSnapshot,          electronRootPids,          electronRootStartTimes,          previous: current.previous,          counters: current.counters,          updatePrevious: input.updatePrevious,        });        const readAt = DateTime.makeUnsafe(merged.sampledAtMs);        const snapshot: ResourceTelemetrySnapshot = {          readAt,          sampleIntervalMs: nativeHealth.sampleIntervalMs,          processes: merged.processes,          groups: merged.groups,          power: Option.match(desktopSnapshot, {            onNone: () => unknownPower(readAt),            onSome: (desktop) => desktop.power,          }),          speedLimitPercent: Option.match(desktopSnapshot, {            onNone: () => Option.none(),            onSome: (desktop) => desktop.speedLimitPercent,          }),          attribution: attributionSnapshot,          health: buildHealth({            native: nativeHealth,            desktop: desktopHealth,            nativeSnapshot,          }),        };        yield* Ref.set(state, {          nativeSnapshot,          desktopSnapshot,          previous: merged.previous,          counters: merged.counters,          latest: snapshot,          lastNativeSequence:            incomingNativeSnapshot?.snapshot.sequence ?? current.lastNativeSequence,          lastNativeGeneration: incomingNativeSnapshot?.generation ?? current.lastNativeGeneration,        });        if (!input.publishWhenLive || (yield* Ref.get(liveState)).retainCount > 0) {          yield* PubSub.publish(changes, snapshot);        }        return snapshot;      }),    );   const ingestNative = (snapshot: NativeTelemetryClient.NativeTelemetrySnapshot) =>    rebuild({ nativeSnapshot: snapshot, updatePrevious: true });  const ingestDesktop = (snapshot: DesktopHostTelemetrySnapshot) =>    Effect.gen(function* () {      const electronRoot = snapshot.electronProcesses.find(        (process) => process.pid === snapshot.electronPid,      );      yield* nativeClient        .setExternalProcesses([          {            pid: snapshot.electronPid,            ...(electronRoot === undefined ? {} : { startTimeMs: electronRoot.creationTimeMs }),          },        ])        .pipe(Effect.ignore);      yield* nativeClient.setHostPowerState(snapshot.power).pipe(Effect.ignore);      return yield* rebuild({        desktopSnapshot: snapshot,        updatePrevious: false,        publishWhenLive: true,      });    });   yield* desktopSubscription.changes.pipe(    Stream.runForEach((snapshot) => ingestDesktop(snapshot)),    Effect.forkScoped,  );   const acquireLive = liveMutex.withPermits(1)(    Effect.uninterruptible(      Effect.gen(function* () {        const current = yield* Ref.get(liveState);        if (current.retainCount > 0) {          yield* Ref.set(liveState, { ...current, retainCount: current.retainCount + 1 });          return;        }         const scope = yield* Scope.make();        yield* desktopReceiver.setDiagnosticsDemand(true).pipe(Effect.ignore);        yield* nativeClient.snapshots.pipe(          Stream.runForEach(ingestNative),          Effect.catch((error) =>            Effect.logWarning("Native resource telemetry stream stopped", {              cause: error.message,            }),          ),          Effect.forkIn(scope),        );        yield* nativeClient.sampleNow.pipe(          Effect.flatMap(ingestNative),          Effect.ignore,          Effect.forkIn(scope),        );        yield* Ref.set(liveState, { retainCount: 1, scope: Option.some(scope) });      }),    ),  );   const releaseLive = liveMutex.withPermits(1)(    Effect.gen(function* () {      const current = yield* Ref.get(liveState);      if (current.retainCount <= 1) {        yield* Ref.set(liveState, { retainCount: 0, scope: Option.none() });        if (Option.isSome(current.scope)) {          yield* Scope.close(current.scope.value, Exit.void).pipe(Effect.ignore);        }        yield* desktopReceiver.setDiagnosticsDemand(false).pipe(Effect.ignore);        return;      }      yield* Ref.set(liveState, { ...current, retainCount: current.retainCount - 1 });    }),  );   const latest = Ref.get(state).pipe(Effect.map((current) => current.latest));  const subscribe = subscribeBeforeSnapshot(    changes,    Effect.acquireRelease(acquireLive, () => releaseLive).pipe(Effect.andThen(latest)),    mutex,  );  const liveChanges = Stream.unwrap(Effect.map(subscribe, ({ changes }) => changes));   const readHistory: ResourceTelemetry["Service"]["readHistory"] = (input) =>    Effect.gen(function* () {      const readAt = yield* DateTime.now;      const normalizedInput = normalizeResourceTelemetryHistoryInput(input);      const historyResult = yield* Effect.result(        nativeClient.readHistory(normalizedInput.windowMs),      );      if (Result.isFailure(historyResult)) {        yield* Effect.logWarning("Failed to read native resource telemetry history", {          cause: historyResult.failure.message,        });      }      const [nativeHealth, desktopHealth] = yield* Effect.all([        nativeClient.health,        desktopReceiver.health,      ]);      const current = yield* Ref.get(state);      return buildResourceTelemetryHistory({        readAt,        windowMs: normalizedInput.windowMs,        bucketMs: normalizedInput.bucketMs,        sampleIntervalMs: nativeHealth.sampleIntervalMs,        serverPid: process.pid,        sidecarPid: Option.map(nativeHealth.hello, (hello) => hello.sidecarPid),        desktopSnapshot: current.desktopSnapshot,        snapshots: Result.isSuccess(historyResult) ? historyResult.success : [],        health: buildHealth({          native: nativeHealth,          desktop: desktopHealth,          nativeSnapshot: current.nativeSnapshot,        }),      });    });  yield* nativeHealthSubscription.changes.pipe(    Stream.runForEach(() => refreshHealth),    Effect.forkScoped,  );  yield* desktopHealthSubscription.changes.pipe(    Stream.runForEach(() => refreshHealth),    Effect.forkScoped,  );   const refresh: ResourceTelemetry["Service"]["refresh"] = nativeClient.sampleNow.pipe(    Effect.flatMap(ingestNative),    Effect.mapError(      (cause) =>        new ResourceTelemetryRefreshFailed({          operation: "refresh",          cause,        }),    ),  );   const validateProcessIdentity: ResourceTelemetry["Service"]["validateProcessIdentity"] = (    identity,  ) =>    nativeClient.sampleNow.pipe(      Effect.map(({ snapshot }) =>        snapshot.processes.some(          (process) => process.pid === identity.pid && process.startTimeMs === identity.startTimeMs,        ),      ),      Effect.mapError(        (cause) =>          new ResourceTelemetryRefreshFailed({            operation: "validateProcessIdentity",            cause,          }),      ),    );   return ResourceTelemetry.of({    latest,    changes: liveChanges,    subscribe,    readHistory,    refresh,    validateProcessIdentity,    retry: nativeClient.retry.pipe(      Effect.zip(Ref.get(state)),      Effect.map(([accepted, current]): ResourceTelemetryRetryResult => ({        accepted,        snapshot: current.latest,      })),    ),  });});

Lines 1–500 of 502. Load more