/**
 * Runbook section 6 — logs are readable and replication is currently healthy.
 *
 * CAP-011 container logs are readable and free of unexpected errors
 *
 * Design note. An earlier draft of this spec allowlisted every WARN it found,
 * which would have permanently hidden `batch ship failing … transport channel
 * closed` — a real replication failure that ran for 115 consecutive seconds
 * after an election. A blanket allowlist is a mute button.
 *
 * So this spec splits the question in two:
 *   1. Is replication healthy RIGHT NOW?  → asserted against a recent window.
 *   2. What has the pod been saying?      → recorded as evidence, and only
 *      genuinely benign, explained lines are tolerated.
 */

import { expect, test } from '@playwright/test';
import { kubectl } from '../support/cluster';
import { observed, recordJson } from '../support/evidence';
import { NAMESPACE, POD_NAMES } from '../support/env';

/** Window that must be quiet for replication to count as currently healthy. */
const HEALTH_WINDOW = '5m';

/** Longer window recorded as evidence, not asserted clean. */
const HISTORY_WINDOW = '30m';

/** Strip ANSI so patterns match; the deployed image colours its output. */
function decolour(text: string): string {
  return text.replace(/\x1b\[[0-9;]*m/g, '');
}

/**
 * Warnings that are genuinely benign on this deployment, each with the reason.
 * An allowlist entry without a rationale is not allowed — if the reason cannot
 * be written down, the line is not understood and must not be silenced.
 */
const BENIGN_WARNINGS: { match: RegExp; why: string }[] = [
  {
    match: /TIDAL_ADMIN_KEY is not set/i,
    why:
      'Stale boot-time WARN. kubelet materialized the projected secret at 05:51:43, after the ' +
      'pod started at 05:41, and the credential poller hot-loaded it. The gate IS live — ' +
      'proven behaviourally in 09-operator-authority.spec.ts. See BUG-003.',
  },
  {
    match: /reading credential file failed.*admin-key/i,
    why:
      'Same cause: the admin-key volume is mounted optional:true, so each poller pass logged a ' +
      'miss until the key existed.',
  },
  {
    match: /TIDAL_CLUSTER_KEY not set/i,
    why: 'Peer authentication uses the internal CA; the shared cluster key is unused here.',
  },
  {
    match: /Multi-process cluster mode enabled/i,
    why: 'Informational notice on every cluster-mode boot — this IS the production HA shape.',
  },
  {
    match: /metrics server bound to non-loopback address/i,
    why:
      'Intentional: :9091 must be reachable by the vmagent scraper. Exposure is contained by ' +
      'the NetworkPolicy, which 04-network-isolation.spec.ts proves refuses foreign pods.',
  },
  {
    match: /catch-up stream open failed; will retry/i,
    why:
      'Self-healing by design — the stream reopens on the next detected gap or retry timer. ' +
      'Tolerated only in the history window; the health window must be clean.',
  },
];

/** Replication distress that must not be ONGOING in the health window. */
const REPLICATION_DISTRESS = /batch ship failing|transport channel closed|quarantin/i;

/** The leader's own "it is fine again" line, which closes a distress burst. */
const SHIP_RECOVERED = /peer recovered; shipping resumed/i;

/*
 * There is deliberately NO consecutive-failure ceiling here.
 *
 * A first version of this test capped `consecutive_failures` at 500, on the guess
 * that a restart burst was "~100" and the 2026-08-30 episode's 2697 was
 * qualitatively different. Measurement killed that idea: this cluster sets
 * `reseed_self_restart: true`, so a node latching a reseed marker exits(0),
 * reinstalls a snapshot, and returns — a ~2 minute absence. At the shipper's
 * 100ms retry cadence that is ~1200-2000 consecutive failures, entirely normal.
 * Observed the same day: tidaldb-2 self-restarted for a shard-1 reseed and
 * tidaldb-0 logged 1950 failures, then `peer recovered … failed_attempts=1950`,
 * with `peer_acked_seqno` back at the leader's frontier.
 *
 * So the COUNT cannot separate "a peer restarted" from "shipping is broken" —
 * both produce large bursts, and the count only encodes how long the peer was
 * away. What separates them is whether the burst CLOSED. That is what this test
 * asserts, and it is strictly stronger: a genuinely stuck shipper never logs
 * `peer recovered`, so it fails here no matter how high or low its count is.
 */

test.describe('section 6 — logs', () => {
  test('replication distress, if any, is a bounded burst that has already recovered', async ({}, testInfo) => {
    const perPod: Record<
      string,
      {
        sampled: number;
        distressLines: string[];
        peakConsecutiveFailures: number;
        recovered: boolean;
        lastDistressIsAfterLastRecovery: boolean;
      }
    > = {};

    for (const pod of POD_NAMES) {
      const result = await observed(testInfo, `recent logs ${pod}`, () =>
        kubectl(['-n', NAMESPACE, 'logs', pod, `--since=${HEALTH_WINDOW}`], {
          timeoutMs: 60_000,
        }),
      );
      expect(result.code, `could not read ${pod} logs: ${result.stderr}`).toBe(0);

      const lines = decolour(result.stdout)
        .split('\n')
        .filter((line) => line.trim() !== '');

      const distressIndexes = lines
        .map((line, index) => (REPLICATION_DISTRESS.test(line) ? index : -1))
        .filter((index) => index >= 0);
      const recoveryIndexes = lines
        .map((line, index) => (SHIP_RECOVERED.test(line) ? index : -1))
        .filter((index) => index >= 0);

      let peak = 0;
      for (const index of distressIndexes) {
        const match = /consecutive_failures=(\d+)/.exec(lines[index]!);
        if (match) peak = Math.max(peak, Number.parseInt(match[1]!, 10));
      }

      const lastDistress = distressIndexes.at(-1) ?? -1;
      const lastRecovery = recoveryIndexes.at(-1) ?? -1;

      perPod[pod] = {
        sampled: lines.length,
        distressLines: distressIndexes.slice(-5).map((index) => lines[index]!),
        peakConsecutiveFailures: peak,
        recovered: recoveryIndexes.length > 0,
        // The load-bearing question: is the newest distress line NEWER than the
        // newest recovery line? If so the burst never closed and shipping is
        // still broken right now.
        lastDistressIsAfterLastRecovery: lastDistress > lastRecovery,
      };
    }

    await recordJson(testInfo, 'replication-health-window', {
      window: HEALTH_WINDOW,
      perPod,
      rationale:
        'A burst that has RECOVERED is designed behavior around a reseed_self_restart ' +
        '(a ~2min absence is ~1200-2000 retries at 100ms). An UNRECOVERED burst means ' +
        'shipping is still broken right now. peakConsecutiveFailures is recorded as ' +
        'evidence only - it measures how long the peer was away, not whether anything ' +
        'is wrong.',
    });

    for (const pod of POD_NAMES) {
      const state = perPod[pod]!;
      if (state.distressLines.length === 0) continue;

      // A pod shipping batches into a closed transport channel is not
      // replicating, even while /cluster/status/local still reports lag=0
      // because the leader has not yet advanced past the stuck position. What
      // distinguishes "a peer just restarted" from "shipping is broken" is
      // whether the burst CLOSED, not whether it happened.
      expect(
        state.lastDistressIsAfterLastRecovery,
        `${pod} replication distress is ONGOING — the newest distress line is newer than ` +
          `the newest 'peer recovered' line, so shipping has not resumed: ` +
          `${state.distressLines.join(' | ')}`,
      ).toBe(false);

    }
  });

  test('no pod logged an ERROR, and every WARN is explained', async ({}, testInfo) => {
    const perPod: Record<
      string,
      { lines: number; errors: string[]; unexplained: string[]; benignCounts: Record<string, number> }
    > = {};

    for (const pod of POD_NAMES) {
      const result = await observed(testInfo, `history logs ${pod}`, () =>
        kubectl(['-n', NAMESPACE, 'logs', pod, `--since=${HISTORY_WINDOW}`, '--tail=400'], {
          timeoutMs: 60_000,
        }),
      );
      expect(result.code, `could not read ${pod} logs: ${result.stderr}`).toBe(0);

      const lines = decolour(result.stdout)
        .split('\n')
        .filter((line) => line.trim() !== '');
      const warnings = lines.filter((line) => /\bWARN\b/.test(line));

      const benignCounts: Record<string, number> = {};
      const unexplained: string[] = [];
      for (const line of warnings) {
        const matched = BENIGN_WARNINGS.find((entry) => entry.match.test(line));
        if (matched) {
          benignCounts[matched.match.source] = (benignCounts[matched.match.source] ?? 0) + 1;
        } else if (REPLICATION_DISTRESS.test(line)) {
          // Recorded, not failed: the previous test owns the health verdict and
          // a historical, recovered episode is legitimate history.
          benignCounts['recovered-replication-episode'] =
            (benignCounts['recovered-replication-episode'] ?? 0) + 1;
        } else {
          unexplained.push(line.slice(0, 200));
        }
      }

      perPod[pod] = {
        lines: lines.length,
        errors: lines.filter((line) => /\bERROR\b/.test(line)).slice(0, 10),
        unexplained: unexplained.slice(0, 10),
        benignCounts,
      };
    }

    await recordJson(testInfo, 'log-classification', {
      window: HISTORY_WINDOW,
      allowlist: BENIGN_WARNINGS.map((entry) => ({ pattern: entry.match.source, why: entry.why })),
      perPod,
    });

    for (const pod of POD_NAMES) {
      expect(perPod[pod].lines, `${pod} should be logging at all`).toBeGreaterThan(0);
      expect(
        perPod[pod].errors,
        `${pod} logged ERROR lines: ${perPod[pod].errors.join(' | ')}`,
      ).toEqual([]);
      expect(
        perPod[pod].unexplained,
        `${pod} logged WARNs with no recorded rationale — investigate and either fix or ` +
          `add an explained allowlist entry: ${perPod[pod].unexplained.join(' | ')}`,
      ).toEqual([]);
    }
  });

  test('the deployed image emits structured uncoloured JSON — level filtering now works in the log store', async ({}, testInfo) => {
    const result = await observed(testInfo, 'log format sample', () =>
      kubectl(['-n', NAMESPACE, 'logs', 'tidaldb-0', '--tail=5'], { timeoutMs: 45_000 }),
    );
    expect(result.code, result.stderr).toBe(0);

    const rawLines = result.stdout.split('\n').filter((line) => line.trim() !== '');
    expect(rawLines.length, 'expected log lines to classify').toBeGreaterThan(0);

    const jsonLines = rawLines.filter((line) => {
      try {
        return typeof JSON.parse(decolour(line)) === 'object';
      } catch {
        return false;
      }
    });
    const ansiLines = rawLines.filter((line) => /\x1b\[/.test(line));

    await recordJson(testInfo, 'log-format', {
      sampled: rawLines.length,
      jsonLines: jsonLines.length,
      ansiLines: ansiLines.length,
      conclusion:
        'Structured uncoloured JSON. BUG-006 (ANSI escapes) stays RESOLVED, and JSON_LOGS ' +
        'went live on m12-harden-20260831 — so VictoriaLogs `level:error` finally matches ' +
        'and filtering no longer has to happen at the source (runbook 9.3).',
    });
    // BUG-006 resolved 2026-08-30. The previous version of this test asserted
    // `ansiLines > 0` — correct for the image running when it was written, and it
    // failed the moment a newer image was rolled. That failure is the test doing
    // its job: it is how the drift got noticed. Now pinned in the other direction
    // so a REGRESSION to coloured output fails here.
    expect(
      ansiLines.length,
      'ANSI escapes are back in container logs — BUG-006 has regressed. Coloured output ' +
        'breaks log-store level matching and makes every downstream filter guess.',
    ).toBe(0);

    // FLIPPED 2026-08-31, on the roll of m12-harden-20260831. This asserted
    // `jsonLines === 0` and instructed its own inversion once structured logging
    // shipped — the feature was implemented all along (logging.rs:85); only the
    // StatefulSet never set JSON_LOGS. Now pinned in the other direction so a
    // REGRESSION to unstructured output fails here, because the whole log-store
    // level taxonomy depends on it.
    expect(
      jsonLines.length,
      'logs stopped being structured JSON — JSON_LOGS has regressed off the StatefulSet. ' +
        'VictoriaLogs `level:` selectors silently match nothing when this breaks, so a ' +
        '"no errors" dashboard becomes indistinguishable from a healthy cluster.',
    ).toBe(rawLines.length);
  });
});