// @effect-diagnostics nodeBuiltinImport:offimport * as NodeFS from "node:fs"; import * as NodeStream from "@effect/platform-node/NodeStream";import {  DesktopHostTelemetryMessage,  type DesktopHostTelemetryMessage as DesktopHostTelemetryMessageValue,  type DesktopHostTelemetrySnapshot,  DesktopTelemetryControlMessage,  type ResourceTelemetrySourceStatus,} from "@t3tools/contracts";import { resolveServerBackgroundActivitySettings } from "@t3tools/shared/backgroundActivitySettings";import * as Context from "effect/Context";import * as DateTime from "effect/DateTime";import * as Duration from "effect/Duration";import * as Effect from "effect/Effect";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 Schema from "effect/Schema";import * as Semaphore from "effect/Semaphore";import * as Scope from "effect/Scope";import * as Stream from "effect/Stream";import * as Ndjson from "effect/unstable/encoding/Ndjson"; import { ServerConfig } from "../config.ts";import { ServerSettingsService } from "../serverSettings.ts";import { subscribeBeforeSnapshotWithoutMutex } from "../utils/subscribeBeforeSnapshot.ts"; const INITIAL_SAMPLE_DEADLINE_MS = 90_000;const MIN_SNAPSHOT_STALE_AFTER_MS = 90_000;const STALE_GRACE_MS = 30_000;const DEFAULT_HOST_POWER_ACTIVE_INTERVAL_MS = 30_000;const DEFAULT_HOST_POWER_IDLE_INTERVAL_MS = 120_000;const STALE_CHECK_INTERVAL = Duration.seconds(30); export class DesktopTelemetryDescriptorUnavailable extends Schema.TaggedErrorClass<DesktopTelemetryDescriptorUnavailable>()(  "DesktopTelemetryDescriptorUnavailable",  {    mode: Schema.String,  },) {  override get message(): string {    return `Desktop telemetry descriptor is unavailable in '${this.mode}' mode.`;  }} export class DesktopTelemetryProtocolMismatch extends Schema.TaggedErrorClass<DesktopTelemetryProtocolMismatch>()(  "DesktopTelemetryProtocolMismatch",  {    expectedVersion: Schema.Number,    receivedVersion: Schema.Number,  },) {  override get message(): string {    return `Desktop telemetry protocol ${this.receivedVersion} is incompatible with expected protocol ${this.expectedVersion}.`;  }} export class DesktopTelemetryDecodeFailed extends Schema.TaggedErrorClass<DesktopTelemetryDecodeFailed>()(  "DesktopTelemetryDecodeFailed",  {    cause: Schema.Defect(),  },) {  override get message(): string {    return "Failed to decode desktop telemetry.";  }} export class DesktopTelemetryStreamFailed extends Schema.TaggedErrorClass<DesktopTelemetryStreamFailed>()(  "DesktopTelemetryStreamFailed",  {    fd: Schema.Number,    cause: Schema.Defect(),  },) {  override get message(): string {    return `Desktop telemetry stream on fd ${this.fd} failed.`;  }} export class DesktopTelemetryStreamClosed extends Schema.TaggedErrorClass<DesktopTelemetryStreamClosed>()(  "DesktopTelemetryStreamClosed",  {    fd: Schema.Number,  },) {  override get message(): string {    return `Desktop telemetry stream on fd ${this.fd} closed.`;  }} export class DesktopTelemetryStale extends Schema.TaggedErrorClass<DesktopTelemetryStale>()(  "DesktopTelemetryStale",  {    fd: Schema.Number,    staleAfterMs: Schema.Number,  },) {  override get message(): string {    return `Desktop telemetry on fd ${this.fd} has not updated for ${this.staleAfterMs}ms.`;  }} export type DesktopTelemetryReceiverError =  | DesktopTelemetryDescriptorUnavailable  | DesktopTelemetryProtocolMismatch  | DesktopTelemetryDecodeFailed  | DesktopTelemetryStreamFailed  | DesktopTelemetryStreamClosed; export class DesktopTelemetryControlFailed extends Schema.TaggedErrorClass<DesktopTelemetryControlFailed>()(  "DesktopTelemetryControlFailed",  {    fd: Schema.Number,    operation: Schema.String,    cause: Schema.Defect(),  },) {  override get message(): string {    return `Desktop telemetry control '${this.operation}' failed on fd ${this.fd}.`;  }} export class DesktopTelemetryControlStalled extends Schema.TaggedErrorClass<DesktopTelemetryControlStalled>()(  "DesktopTelemetryControlStalled",  {    fd: Schema.Number,    remainingBytes: Schema.Number,  },) {  override get message(): string {    return `Desktop telemetry control stalled on fd ${this.fd} with ${this.remainingBytes} bytes remaining.`;  }} export type DesktopTelemetryControlError =  | DesktopTelemetryControlFailed  | DesktopTelemetryControlStalled; export interface DesktopTelemetryReceiverHealth {  readonly status: ResourceTelemetrySourceStatus;  readonly lastSampleAt: Option.Option<DateTime.Utc>;  readonly lastError: Option.Option<string>;} export class DesktopTelemetryReceiver extends Context.Service<  DesktopTelemetryReceiver,  {    readonly latest: Effect.Effect<Option.Option<DesktopHostTelemetrySnapshot>>;    readonly changes: Stream.Stream<DesktopHostTelemetrySnapshot>;    readonly subscribe: Effect.Effect<      {        readonly latest: Option.Option<DesktopHostTelemetrySnapshot>;        readonly changes: Stream.Stream<DesktopHostTelemetrySnapshot>;      },      never,      Scope.Scope    >;    readonly health: Effect.Effect<DesktopTelemetryReceiverHealth>;    readonly subscribeHealth: Effect.Effect<      {        readonly latest: DesktopTelemetryReceiverHealth;        readonly changes: Stream.Stream<DesktopTelemetryReceiverHealth>;      },      never,      Scope.Scope    >;    readonly setDiagnosticsDemand: (      enabled: boolean,    ) => Effect.Effect<void, DesktopTelemetryControlError>;  }>()("t3/resourceTelemetry/DesktopTelemetryReceiver") {} const decodeMessage = Schema.decodeUnknownEffect(DesktopHostTelemetryMessage);const encodeControlMessage = Schema.encodeEffect(  Schema.fromJsonString(DesktopTelemetryControlMessage),);const isDescriptorUnavailable = Schema.is(DesktopTelemetryDescriptorUnavailable);const isProtocolMismatch = Schema.is(DesktopTelemetryProtocolMismatch);const isDecodeFailed = Schema.is(DesktopTelemetryDecodeFailed);const isStreamFailed = Schema.is(DesktopTelemetryStreamFailed); export function isDesktopTelemetryContactStale(  lastContactAtMs: Option.Option<number>,  nowMs: number,): boolean {  return Option.exists(    lastContactAtMs,    (lastContact) => nowMs - lastContact >= INITIAL_SAMPLE_DEADLINE_MS,  );} export function resolveDesktopTelemetrySnapshotStaleAfterMs(  activeIntervalMs: number,  idleIntervalMs: number,): number {  return Math.max(    MIN_SNAPSHOT_STALE_AFTER_MS,    Math.max(activeIntervalMs, idleIntervalMs) + STALE_GRACE_MS,  );} export function initialDesktopTelemetryContactAt(  desktopTelemetryFd: number | undefined,  nowMs: number,): Option.Option<number> {  return desktopTelemetryFd === undefined ? Option.none() : Option.some(nowMs);} export const recordDesktopTelemetrySampleHealth = Effect.fn(  "resourceTelemetry.desktopTelemetryReceiver.recordSampleHealth",)(function* (  health: Ref.Ref<DesktopTelemetryReceiverHealth>,  healthChanges: PubSub.PubSub<DesktopTelemetryReceiverHealth>,  sampledAt: DateTime.Utc,) {  const next: DesktopTelemetryReceiverHealth = {    status: "healthy",    lastSampleAt: Option.some(sampledAt),    lastError: Option.none(),  };  yield* Ref.set(health, next);  yield* PubSub.publish(healthChanges, next);}); function normalizeReceiverError(error: unknown): DesktopTelemetryReceiverError {  if (    isDescriptorUnavailable(error) ||    isProtocolMismatch(error) ||    isDecodeFailed(error) ||    isStreamFailed(error)  ) {    return error;  }  return new DesktopTelemetryDecodeFailed({ cause: error });} function messageVersion(value: unknown): number | undefined {  if (typeof value !== "object" || value === null) return undefined;  const version = Reflect.get(value, "version");  return typeof version === "number" ? version : undefined;} export const writeAllToFileDescriptor = Effect.fn(  "resourceTelemetry.desktopTelemetryReceiver.writeAllToFileDescriptor",)(function* (fd: number, payload: Buffer) {  let offset = 0;  while (offset < payload.byteLength) {    const written = yield* Effect.callback<number, DesktopTelemetryControlFailed>(      (resume, signal) => {        if (signal.aborted) return;        try {          NodeFS.write(            fd,            payload,            offset,            payload.byteLength - offset,            null,            (error, bytesWritten) => {              if (error) {                resume(                  Effect.fail(                    new DesktopTelemetryControlFailed({                      fd,                      operation: "write",                      cause: error,                    }),                  ),                );                return;              }              resume(Effect.succeed(bytesWritten));            },          );        } catch (cause) {          resume(            Effect.fail(              new DesktopTelemetryControlFailed({                fd,                operation: "write",                cause,              }),            ),          );        }      },    );    yield* requireDesktopTelemetryWriteProgress(fd, payload.byteLength - offset, written);    offset += written;  }}); export function requireDesktopTelemetryWriteProgress(  fd: number,  remainingBytes: number,  written: number,): Effect.Effect<void, DesktopTelemetryControlStalled> {  return written > 0    ? Effect.void    : Effect.fail(new DesktopTelemetryControlStalled({ fd, remainingBytes }));} export const make = Effect.fn("resourceTelemetry.desktopTelemetryReceiver.make")(function* () {  const config = yield* ServerConfig;  const serverSettings = yield* ServerSettingsService;  const latest = yield* Ref.make(Option.none<DesktopHostTelemetrySnapshot>());  const receiverStartedAt = yield* DateTime.now;  const lastContactAtMs = yield* Ref.make(    initialDesktopTelemetryContactAt(      config.desktopTelemetryFd,      DateTime.toEpochMillis(receiverStartedAt),    ),  );  const snapshotStaleAfterMs = yield* Ref.make(    resolveDesktopTelemetrySnapshotStaleAfterMs(      DEFAULT_HOST_POWER_ACTIVE_INTERVAL_MS,      DEFAULT_HOST_POWER_IDLE_INTERVAL_MS,    ),  );  const changes = yield* PubSub.sliding<DesktopHostTelemetrySnapshot>(8);  const healthChanges = yield* PubSub.sliding<DesktopTelemetryReceiverHealth>(4);  const controlMutex = yield* Semaphore.make(1);  const snapshotMutex = yield* Semaphore.make(1);  const health = yield* Ref.make<DesktopTelemetryReceiverHealth>({    status: config.desktopTelemetryFd === undefined ? "unavailable" : "starting",    lastSampleAt: Option.none(),    lastError:      config.desktopTelemetryFd === undefined        ? Option.some(            new DesktopTelemetryDescriptorUnavailable({              mode: config.mode,            }).message,          )        : Option.none(),  });  const updateHealth = (    update: (current: DesktopTelemetryReceiverHealth) => DesktopTelemetryReceiverHealth,  ) =>    Ref.modify(health, (current) => {      const next = update(current);      return [next, next];    }).pipe(      Effect.flatMap((next) => PubSub.publish(healthChanges, next)),      Effect.asVoid,    );  const updateSampleHealth = (sampledAt: DateTime.Utc) =>    recordDesktopTelemetrySampleHealth(health, healthChanges, sampledAt);   const sendControlMessage = (message: DesktopTelemetryControlMessage) =>    controlMutex.withPermits(1)(      Effect.gen(function* () {        const fd = config.desktopTelemetryControlFd;        if (fd === undefined) return;        const encoded = yield* encodeControlMessage(message).pipe(          Effect.mapError(            (cause) =>              new DesktopTelemetryControlFailed({                fd,                operation: "encode",                cause,              }),          ),        );        yield* writeAllToFileDescriptor(fd, Buffer.from(`${encoded}\n`)).pipe(          Effect.tapError((error) =>            updateHealth((current) => ({              ...current,              status: "degraded",              lastError: Option.some(error.message),            })),          ),        );      }),    );  const setDiagnosticsDemand: DesktopTelemetryReceiver["Service"]["setDiagnosticsDemand"] = (    enabled,  ) =>    sendControlMessage({      version: 1,      type: "setDiagnosticsDemand",      enabled,    });   const sendHostPowerIntervals = (    settings: Parameters<typeof resolveServerBackgroundActivitySettings>[0],  ) => {    const resolved = resolveServerBackgroundActivitySettings(settings);    const activeIntervalMs = Math.max(      1,      Math.round(Duration.toMillis(resolved.hostPowerMonitorActiveInterval)),    );    const idleIntervalMs = Math.max(      1,      Math.round(Duration.toMillis(resolved.hostPowerMonitorIdleInterval)),    );    return sendControlMessage({      version: 1,      type: "setHostPowerIntervals",      activeIntervalMs,      idleIntervalMs,    }).pipe(      Effect.andThen(        Ref.set(          snapshotStaleAfterMs,          resolveDesktopTelemetrySnapshotStaleAfterMs(activeIntervalMs, idleIntervalMs),        ),      ),    );  };  if (config.desktopTelemetryControlFd !== undefined) {    const settingsChanges = yield* serverSettings.subscribeChanges;    const settings = yield* serverSettings.getSettings;    yield* sendHostPowerIntervals(settings).pipe(      Effect.catchCause((cause) =>        Effect.logWarning("Failed to configure desktop host-power intervals", {          cause: String(cause),        }),      ),    );    yield* settingsChanges.pipe(      Stream.runForEach((settings) =>        sendHostPowerIntervals(settings).pipe(          Effect.catchCause((cause) =>            Effect.logWarning("Failed to update desktop host-power intervals", {              cause: String(cause),            }),          ),        ),      ),      Effect.forkScoped,    );  }   if (config.desktopTelemetryFd !== undefined) {    const fd = config.desktopTelemetryFd;    const readable = yield* Effect.acquireRelease(      Effect.try({        try: () =>          NodeFS.createReadStream("", {            fd,            autoClose: true,          }),        catch: (cause) => new DesktopTelemetryStreamFailed({ fd, cause }),      }),      (stream) =>        Effect.sync(() => {          stream.destroy();        }),    );     const messages: Stream.Stream<DesktopHostTelemetryMessageValue, DesktopTelemetryReceiverError> =      NodeStream.fromReadable<Uint8Array, DesktopTelemetryStreamFailed>({        evaluate: () => readable,        closeOnDone: true,        onError: (cause) => new DesktopTelemetryStreamFailed({ fd, cause }),      }).pipe(        Stream.pipeThroughChannel(Ndjson.decode({ ignoreEmptyLines: true })),        Stream.mapEffect(          (            value,          ): Effect.Effect<            DesktopHostTelemetryMessageValue,            DesktopTelemetryProtocolMismatch | DesktopTelemetryDecodeFailed          > => {            const version = messageVersion(value);            if (version !== undefined && version !== 1) {              return Effect.fail(                new DesktopTelemetryProtocolMismatch({                  expectedVersion: 1,                  receivedVersion: version,                }),              );            }            return decodeMessage(value).pipe(              Effect.mapError((cause) => new DesktopTelemetryDecodeFailed({ cause })),            );          },        ),        Stream.mapError(normalizeReceiverError),      );     yield* messages.pipe(      Stream.runForEach((message) => {        const recordContact = DateTime.now.pipe(          Effect.flatMap((now) =>            Ref.set(lastContactAtMs, Option.some(DateTime.toEpochMillis(now))),          ),        );        if (message.type === "desktopTelemetryHello") {          return recordContact.pipe(            Effect.andThen(              updateHealth(                (current): DesktopTelemetryReceiverHealth => ({                  ...current,                  status: "healthy",                  lastError: Option.none(),                }),

Lines 1–500 of 662. Load more