import assert from "node:assert/strict";
import { test } from "bun:test";

import { nodeWebObserveRunnerKafkaDebugReplaySource } from "./hwlab-node-web-observe-runner-kafka-debug-replay-source";
import { nodeWebObserveRunnerSource } from "./hwlab-node-web-observe-runner-source";

test("Workbench Kafka debug replay source parses the YAML contract and reducer counts", () => {
  const source = nodeWebObserveRunnerKafkaDebugReplaySource();
  const build = new Function(`${source}\nreturn { parseWorkbenchKafkaDebugReplayConfig, workbenchKafkaDebugReplayCounts, workbenchKafkaDebugReplayLayerCounts, workbenchKafkaDebugReplayRawCounts, workbenchKafkaDebugReplayTypedOutcome, workbenchKafkaDebugReplayValidateLayerCounts, workbenchKafkaDebugReplayValidateTraceTimeline, workbenchKafkaDebugReplayValidateRawWindow };`) as () => {
    parseWorkbenchKafkaDebugReplayConfig: (raw: string) => Record<string, unknown>;
    workbenchKafkaDebugReplayCounts: (text: string) => Record<string, number> | null;
    workbenchKafkaDebugReplayLayerCounts: (text: string) => Record<string, unknown> | null;
    workbenchKafkaDebugReplayRawCounts: (text: string) => Record<string, unknown> | null;
    workbenchKafkaDebugReplayTypedOutcome: (input: Record<string, unknown>) => Record<string, unknown>;
    workbenchKafkaDebugReplayValidateLayerCounts: (counts: Record<string, unknown> | null, evidence: Record<string, unknown>) => void;
    workbenchKafkaDebugReplayValidateTraceTimeline: (timeline: Record<string, unknown>, evidence: Record<string, unknown>) => void;
    workbenchKafkaDebugReplayValidateRawWindow: (window: Record<string, unknown>, evidence: Record<string, unknown>) => void;
  };
  const helpers = build();
  assert.deepEqual(helpers.parseWorkbenchKafkaDebugReplayConfig(JSON.stringify({
    enabled: true,
    topic: "hwlab.event.debug.v1",
    groupPrefix: "hwlab-v03-workbench-isolated-debug",
    completionTimeoutMs: 30_000,
    productEventsPath: "/v1/workbench/events",
    debugEventsPath: "/v1/workbench/debug/kafka-sse/events",
    requestObservationQuietMs: 3_000,
  })), {
    enabled: true,
    topic: "hwlab.event.debug.v1",
    groupPrefix: "hwlab-v03-workbench-isolated-debug",
    completionTimeoutMs: 30_000,
    productEventsPath: "/v1/workbench/events",
    debugEventsPath: "/v1/workbench/debug/kafka-sse/events",
    requestObservationQuietMs: 3_000,
    valuesRedacted: true,
  });
  assert.deepEqual(helpers.workbenchKafkaDebugReplayCounts("35/35 applied"), { appliedCount: 35, receivedCount: 35 });
  assert.equal(helpers.workbenchKafkaDebugReplayCounts("等待重放"), null);
  assert.deepEqual(helpers.workbenchKafkaDebugReplayLayerCounts(
    "server scanned=27 · matched=2 · delivered=2 client local overlay received=2 · decoded=2 · applied=2",
  ), {
    server: { scanned: 27, matched: 2, delivered: 2 },
    client: { received: 2, decoded: 2, applied: 2 },
    valuesRedacted: true,
  });
  assert.deepEqual(helpers.workbenchKafkaDebugReplayRawCounts(
    "received=34 · retained=30 · decoded=33 · rejected=1 · evicted=4 · bytes=8192",
  ), {
    received: 34,
    retained: 30,
    decoded: 33,
    rejected: 1,
    evicted: 4,
    bytes: 8192,
    valuesRedacted: true,
  });
  assert.deepEqual(helpers.workbenchKafkaDebugReplayTypedOutcome({ status: "completed", layerCounts: { server: {} } }), {
    phase: "terminal",
    code: "terminal_complete",
    source: "typed-dom-v2",
    valuesRedacted: true,
  });
  assert.deepEqual(helpers.workbenchKafkaDebugReplayTypedOutcome({ status: "incomplete", errorText: "records_scanned_but_filter_mismatch: no replayId match" }), {
    phase: "server",
    code: "records_scanned_but_filter_mismatch",
    source: "legacy-dom",
    valuesRedacted: true,
  });
  const evidence = {
    phase: "terminal",
    code: "terminal_complete",
    traceId: "trc_fixture",
    legacyCounts: { receivedCount: 2, appliedCount: 2 },
  };
  assert.throws(
    () => helpers.workbenchKafkaDebugReplayValidateLayerCounts(null, evidence),
    /typed server\/client count contract is missing/u,
  );
  assert.throws(
    () => helpers.workbenchKafkaDebugReplayValidateTraceTimeline({ authorityAvailable: false }, evidence),
    /source-authority contract is missing/u,
  );
  assert.throws(
    () => helpers.workbenchKafkaDebugReplayValidateRawWindow({ available: false }, evidence),
    /raw HWLAB Event window contract is missing/u,
  );
  assert.throws(
    () => helpers.workbenchKafkaDebugReplayValidateRawWindow({
      available: true,
      samePage: true,
      sourceContractPresent: true,
      unfilteredContractPresent: true,
      productEventSourceRequestCount: 0,
      textPresent: false,
      textBytes: 0,
      counts: { received: 0, retained: 0, decoded: 0, rejected: 0, evicted: 0, bytes: 0 },
    }, evidence),
    /did not retain a readable product SSE frame/u,
  );
  assert.doesNotThrow(() => helpers.workbenchKafkaDebugReplayValidateLayerCounts({
    server: { scanned: 27, matched: 2, delivered: 2 },
    client: { received: 2, decoded: 2, applied: 2 },
  }, evidence));
  assert.doesNotThrow(() => helpers.workbenchKafkaDebugReplayValidateTraceTimeline({
    authorityAvailable: true,
    sourceAuthorityRowCount: 2,
    readableSourceRowCount: 2,
  }, evidence));
  assert.doesNotThrow(() => helpers.workbenchKafkaDebugReplayValidateRawWindow({
    available: true,
    samePage: true,
    sourceContractPresent: true,
    unfilteredContractPresent: true,
    productEventSourceRequestCount: 0,
    textPresent: true,
    textBytes: 256,
    counts: { received: 2, retained: 2, decoded: 2, rejected: 0, evicted: 0, bytes: 256 },
  }, evidence));
});

test("generated observer runner contains the typed debug replay command and passes node syntax check", async () => {
  const source = nodeWebObserveRunnerSource();
  assert.match(source, /validateWorkbenchKafkaDebugReplay/u);
  assert.match(source, /workbench-kafka-debug-toggle/u);
  assert.match(source, /workbench-kafka-debug-replay/u);
  assert.match(source, /workbench-kafka-debug-counts/u);
  assert.match(source, /workbench-kafka-debug-layer-counts/u);
  assert.match(source, /workbenchKafkaDebugReplayWaitForStablePanel/u);
  assert.match(source, /layerDomStabilization/u);
  assert.match(source, /textHash/u);
  assert.match(source, /workbench-kafka-debug-tab-raw/u);
  assert.match(source, /workbench-raw-hwlab-counts/u);
  assert.match(source, /data-event-seq-authority/u);
  const child = Bun.spawn(["node", "--input-type=module", "--check", "-"], { stdin: "pipe", stdout: "pipe", stderr: "pipe" });
  child.stdin.write(source);
  child.stdin.end();
  const [stderr, exitCode] = await Promise.all([new Response(child.stderr).text(), child.exited]);
  assert.equal(exitCode, 0, stderr);
});