import type {
  DesktopHostTelemetrySnapshot,
  ResourceMonitorProcessSample,
  ResourceMonitorSnapshotEvent,
} from "@t3tools/contracts";
import { describe, expect, it } from "@effect/vitest";
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 Stream from "effect/Stream";
import * as TestClock from "effect/testing/TestClock";

import * as DesktopTelemetryReceiver from "./DesktopTelemetryReceiver.ts";
import * as NativeTelemetryClient from "./NativeTelemetryClient.ts";
import * as ResourceAttribution from "./ResourceAttribution.ts";
import * as ResourceTelemetry from "./ResourceTelemetry.ts";

function processSample(
  input: Partial<ResourceMonitorProcessSample> &
    Pick<ResourceMonitorProcessSample, "pid" | "ppid" | "startTimeMs">,
): ResourceMonitorProcessSample {
  return {
    runTimeMs: 1_000,
    name: `process-${input.pid}`,
    command: `process-${input.pid}`,
    status: "Running",
    cpuPercent: 0,
    cpuTimeMs: 0,
    residentBytes: 1_024,
    virtualBytes: 2_048,
    ioReadBytes: 0,
    ioWriteBytes: 0,
    ioSemantics: "storage",
    ...input,
  };
}

function nativeSnapshot(input: {
  readonly sequence: number;
  readonly sampledAtUnixMs: number;
  readonly childCpuTimeMs: number;
  readonly childWriteBytes: number;
}): ResourceMonitorSnapshotEvent {
  const processes = [
    processSample({
      pid: process.pid,
      ppid: 1,
      startTimeMs: 100,
      cpuTimeMs: input.sequence * 10,
    }),
    processSample({
      pid: 4_242,
      ppid: process.pid,
      startTimeMs: 200,
      name: "codex",
      command: "codex app-server",
      cpuTimeMs: input.childCpuTimeMs,
      ioWriteBytes: input.childWriteBytes,
    }),
    processSample({
      pid: 5_000,
      ppid: 1,
      startTimeMs: 300,
      name: "electron",
      command: "electron",
      cpuTimeMs: input.sequence * 20,
    }),
    processSample({
      pid: 9_000,
      ppid: process.pid,
      startTimeMs: 400,
      name: "t3-resource-monitor",
      command: "t3-resource-monitor",
      cpuTimeMs: input.sequence * 5,
    }),
  ];
  return {
    version: 2,
    type: "snapshot",
    sequence: input.sequence,
    sampledAtUnixMs: input.sampledAtUnixMs,
    collectionDurationMicros: 300,
    scannedProcessCount: 80,
    retainedProcessCount: processes.length,
    inaccessibleProcessCount: 1,
    processes,
  };
}

function desktopSnapshot(sampledAtUnixMs: number): DesktopHostTelemetrySnapshot {
  const sampledAt = DateTime.makeUnsafe(sampledAtUnixMs);
  return {
    version: 1,
    type: "desktopTelemetry",
    sequence: 1,
    sampledAtUnixMs,
    electronPid: 5_000,
    power: {
      source: "electron-main",
      idle: "false",
      idleSeconds: 2,
      locked: "false",
      suspended: false,
      onBattery: "true",
      lowPowerMode: "unknown",
      thermalState: "fair",
      stale: false,
      updatedAt: sampledAt,
    },
    speedLimitPercent: Option.some(90),
    electronProcesses: [
      {
        pid: 5_000,
        creationTimeMs: 300,
        type: "Browser",
        name: "electron",
        cpuPercent: 2,
        cumulativeCpuSeconds: 0.02,
        idleWakeupsPerSecond: 3,
        workingSetBytes: 4_096,
        peakWorkingSetBytes: 8_192,
      },
    ],
  };
}

describe("ResourceTelemetry", () => {
  it.effect("enables live native and Electron collection only while changes are retained", () =>
    Effect.gen(function* () {
      const sampledAtUnixMs = DateTime.toEpochMillis(yield* DateTime.now);
      const sample = nativeSnapshot({
        sequence: 1,
        sampledAtUnixMs,
        childCpuTimeMs: 100,
        childWriteBytes: 1_000,
      });
      const demandChanges = yield* Ref.make<ReadonlyArray<boolean>>([]);
      const nativeLayer = NativeTelemetryClient.layerTest({
        sampleNow: Effect.succeed(sample),
        health: Effect.succeed({
          status: "healthy",
          hello: Option.none(),
          lastSampleAt: Option.none(),
          lastError: Option.none(),
          restartCount: 0,
          sampleIntervalMs: 1_000,
        }),
      });
      const desktopLayer = DesktopTelemetryReceiver.layerTest({
        latest: Effect.succeedSome(desktopSnapshot(sampledAtUnixMs)),
        setDiagnosticsDemand: (enabled) =>
          Ref.update(demandChanges, (changes) => [...changes, enabled]),
      });
      const telemetryLayer = ResourceTelemetry.layer.pipe(
        Layer.provide(Layer.mergeAll(nativeLayer, desktopLayer, ResourceAttribution.layer)),
      );

      const live = yield* Stream.runHead(
        Effect.gen(function* () {
          const telemetry = yield* ResourceTelemetry.ResourceTelemetry;
          return telemetry.changes;
        }).pipe(Stream.unwrap),
      ).pipe(Effect.provide(telemetryLayer));

      expect(Option.isSome(live)).toBe(true);
      expect(yield* Ref.get(demandChanges)).toEqual([true, false]);
    }),
  );

  it.effect("combines native, Electron, attribution, retry, and history data", () =>
    Effect.gen(function* () {
      const startedAt = DateTime.toEpochMillis(yield* DateTime.now);
      const samples = [
        nativeSnapshot({
          sequence: 1,
          sampledAtUnixMs: startedAt,
          childCpuTimeMs: 100,
          childWriteBytes: 1_000,
        }),
        nativeSnapshot({
          sequence: 2,
          sampledAtUnixMs: startedAt + 1_000,
          childCpuTimeMs: 350,
          childWriteBytes: 5_000,
        }),
        nativeSnapshot({
          sequence: 1,
          sampledAtUnixMs: startedAt + 2_000,
          childCpuTimeMs: 500,
          childWriteBytes: 7_000,
        }),
      ] as const;
      const sampleIndex = yield* Ref.make(0);
      const externalProcesses = yield* Ref.make<
        ReadonlyArray<{ readonly pid: number; readonly startTimeMs?: number }>
      >([]);
      const retryCount = yield* Ref.make(0);
      const nativeHealth = yield* Ref.make<NativeTelemetryClient.NativeTelemetryClientHealth>({
        status: "healthy",
        hello: Option.some({
          version: 2,
          type: "hello",
          sidecarVersion: "0.1.0",
          sidecarPid: 9_000,
          platform: "test",
          arch: "test",
          capabilities: {
            cumulativeCpuTime: true,
            currentCpuPercent: true,
            residentMemory: true,
            virtualMemory: true,
            ioBytes: true,
            processStartTime: true,
            processTree: true,
          },
        }),
        lastSampleAt: Option.some(DateTime.makeUnsafe(startedAt)),
        lastError: Option.none(),
        restartCount: 2,
        sampleIntervalMs: 1_000,
      });
      const nativeHealthChanges =
        yield* PubSub.sliding<NativeTelemetryClient.NativeTelemetryClientHealth>(4);
      const nativeLayer = NativeTelemetryClient.layerTest({
        setExternalProcesses: (processes) => Ref.set(externalProcesses, processes),
        readHistory: () => Effect.succeed(samples.slice(0, 2)),
        sampleNow: Ref.modify(sampleIndex, (index) => [
          samples[Math.min(index, samples.length - 1)]!,
          index + 1,
        ]),
        retry: Ref.updateAndGet(retryCount, (count) => count + 1).pipe(Effect.as(true)),
        health: Ref.get(nativeHealth),
        healthChanges: Stream.fromPubSub(nativeHealthChanges),
      });
      const desktopLayer = DesktopTelemetryReceiver.layerTest({
        latest: Effect.succeedSome(desktopSnapshot(startedAt)),
        health: Effect.succeed({
          status: "healthy",
          lastSampleAt: Option.some(DateTime.makeUnsafe(startedAt)),
          lastError: Option.none(),
        }),
      });
      const attributionLayer = ResourceAttribution.layer;
      const dependencies = Layer.mergeAll(nativeLayer, desktopLayer, attributionLayer);
      const telemetryLayer = ResourceTelemetry.layer.pipe(Layer.provide(dependencies));
      const layer = Layer.mergeAll(dependencies, telemetryLayer);

      yield* Effect.gen(function* () {
        const telemetry = yield* ResourceTelemetry.ResourceTelemetry;
        const attribution = yield* ResourceAttribution.ResourceAttribution;

        expect(yield* Ref.get(externalProcesses)).toEqual([{ pid: 5_000 }]);

        yield* attribution.record({
          component: "provider-event-log",
          operation: "append",
          logicalWriteBytes: 512,
          count: 2,
          durationMs: 4,
        });
        const first = yield* telemetry.refresh;
        expect(first.groups.backend.processCount).toBe(2);
        expect(first.groups.electron.processCount).toBe(1);
        expect(first.groups.monitor.processCount).toBe(1);
        expect(first.power.onBattery).toBe("true");
        expect(Option.getOrNull(first.speedLimitPercent)).toBe(90);
        expect(first.attribution.entries).toEqual([
          {
            component: "provider-event-log",
            operation: "append",
            logicalReadBytes: 0,
            logicalWriteBytes: 512,
            count: 2,
            durationMs: 4,
          },
        ]);

        yield* TestClock.adjust(Duration.seconds(1));
        const second = yield* telemetry.refresh;
        const codex = second.processes.find((entry) => entry.identity.pid === 4_242);
        expect(codex?.cpuPercent).toBe(25);
        expect(codex?.ioWriteBytesPerSecond).toBe(4_000);
        expect(second.groups.backend.ioWriteBytes).toBe(4_000);
        expect(second.health.collectionDurationMicros).toBe(300);
        expect(second.health.scannedProcessCount).toBe(80);
        expect(second.health.inaccessibleProcessCount).toBe(1);

        const history = yield* telemetry.readHistory({
          windowMs: 60_000,
          bucketMs: 10_000,
        });
        expect(history.retainedSampleCount).toBeGreaterThan(0);
        expect(
          history.topProcesses.find((entry) => entry.identity.pid === 4_242)?.sampleCount,
        ).toBe(2);
        expect(history.topProcesses.find((entry) => entry.identity.pid === 4_242)?.cpuTimeMs).toBe(
          250,
        );
        expect(
          history.topProcesses.find((entry) => entry.identity.pid === 4_242)?.ioWriteBytes,
        ).toBe(4_000);
        expect(history.buckets.reduce((total, bucket) => total + bucket.ioWriteBytes, 0)).toBe(
          4_000,
        );

        const retry = yield* telemetry.retry;
        expect(retry.accepted).toBe(true);
        expect(yield* Ref.get(retryCount)).toBe(1);

        yield* Ref.update(nativeHealth, (current) => ({
          ...current,
          hello: Option.map(current.hello, (hello) => ({
            ...hello,
            sidecarPid: 9_001,
          })),
          restartCount: 3,
        }));
        yield* TestClock.adjust(Duration.seconds(1));
        const restarted = yield* telemetry.refresh;
        expect(DateTime.toEpochMillis(restarted.readAt)).toBe(startedAt + 2_000);
        expect(Option.getOrNull(restarted.health.sidecarPid)).toBe(9_001);

        yield* Ref.update(nativeHealth, (current) => ({
          ...current,
          status: "degraded" as const,
          lastError: Option.some("collector exited"),
        }));
        yield* PubSub.publish(nativeHealthChanges, yield* Ref.get(nativeHealth));
        yield* Effect.yieldNow;
        const healthUpdate = yield* telemetry.latest;
        expect(healthUpdate.health.native.status).toBe("degraded");
        expect(Option.getOrNull(healthUpdate.health.native.lastError)).toBe("collector exited");
        const degradedHistory = yield* telemetry.readHistory({
          windowMs: 60_000,
          bucketMs: 10_000,
        });
        expect(degradedHistory.health.native.status).toBe("degraded");
      }).pipe(Effect.provide(layer));
    }),
  );
});