/**
 * Demo capture — the stakeholder walkthrough of the deploy verification.
 *
 * Every capture in here re-runs the SAME assertion the regression suite makes,
 * then photographs the asserted state. Nothing is captured that has not just
 * been proven in the same test: if the assertion fails, no image is written.
 * That is the whole contract — Playwright establishes truth, Remotion presents
 * it, and this file is the seam.
 *
 * Beats map to demo/storyboard.md and capability IDs to
 * demo/capability-inventory.md.
 */

import { expect, test } from '@playwright/test';
import { kubectl, run, tidalctl, withPortForward } from '../../support/cluster';
import {
  BACKUP_NAMESPACE,
  BACKUP_SCHEDULE,
  DASHBOARD_UID,
  FOREIGN_NAMESPACE,
  FOREIGN_POD,
  NAMESPACE,
  OBS_NAMESPACE,
  POD_NAMES,
  PORT_CLIENT,
  PORT_METRICS,
  PUBLIC_BASE_URL,
  PUBLIC_HOST,
  adminKey,
  apiKey,
  grafanaPassword,
} from '../../support/env';
import {
  CAPTURE_HEIGHT,
  CAPTURE_WIDTH,
  captureProofPanel,
  recordScreenshot,
  writeManifestFragment,
  type CaptureRecord,
} from '../support/proof-panel';

const OPERATOR = ['cluster operator'];
const records: CaptureRecord[] = [];

test.afterAll(async () => {
  await writeManifestFragment('deploy-verification', records);
});

test.describe('demo capture — deploy verification', () => {
  test('CAP-002 every node is converged', async ({ page, playwright }, testInfo) => {
    const key = apiKey();
    const rows: string[] = [];

    for (const pod of POD_NAMES) {
      const status = await withPortForward(NAMESPACE, pod, PORT_CLIENT, async (forward) => {
        const context = await playwright.request.newContext({
          ignoreHTTPSErrors: true,
          extraHTTPHeaders: { authorization: `Bearer ${key}` },
        });
        try {
          const response = await context.get(
            `https://127.0.0.1:${forward.localPort}/cluster/status/local`,
          );
          expect(response.status()).toBe(200);
          return await response.json();
        } finally {
          await context.dispose();
        }
      });

      // Assert before capturing. A capture is only ever a photograph of an
      // already-proven state.
      expect(status.reseed_required, `${pod} reseed`).toBe(false);
      for (const shard of status.shards) {
        expect(shard.lag_events, `${pod} shard ${shard.shard} lag`).toBe(0);
      }

      const shards = status.shards
        .map(
          (s: { shard: number; applied_events: number; lag_events: number; leader: string }) =>
            `group ${s.shard}: applied=${s.applied_events}  lag=${s.lag_events}  leader=${s.leader}`,
        )
        .join('\n           ');
      rows.push(`${status.region}   reseed=${status.reseed_required}\n           ${shards}`);
    }

    records.push(
      await captureProofPanel(
        page,
        testInfo,
        {
          captureId: 'CAP-002-convergence',
          capabilityId: 'CAP-002',
          title: 'Every node agrees, and none is behind',
          subtitle:
            'Each node is asked for its own view. The aggregated endpoint under-reports peers, ' +
            'so the per-node answer is the authoritative one.',
          blocks: [
            {
              command: 'GET /cluster/status/local  (each pod, via port-forward)',
              output: rows.join('\n\n'),
              verdict:
                'lag=0 on all 3 shard groups of all 3 nodes, no reseed pending, one agreed leader per group.',
            },
          ],
          footnote:
            'A pod can be Ready while its replication is stalled — this is the check the ' +
            '2026-08-20 reseed livelock defeated.',
        },
        {
          expected: 'Three nodes, zero lag on every shard group, no reseed pending',
          businessPurpose:
            'Quorum with one-node fault tolerance actually exists, rather than being assumed from pod readiness',
          personas: OPERATOR,
        },
      ),
    );
  });

  test('CAP-005 CAP-006 the boundary refuses, then a quorum write commits', async ({
    page,
    playwright,
  }, testInfo) => {
    const anonymous = await playwright.request.newContext();
    const wrong = await playwright.request.newContext({
      extraHTTPHeaders: { authorization: 'Bearer definitely-not-the-key' },
    });
    const authorized = await playwright.request.newContext({
      extraHTTPHeaders: { authorization: `Bearer ${apiKey()}` },
    });

    try {
      const target = `${PUBLIC_BASE_URL}/search?query=verification&limit=1`;
      const none = await anonymous.get(target);
      const bad = await wrong.get(target);
      const good = await authorized.get(target);

      expect(none.status()).toBe(401);
      expect(bad.status()).toBe(401);
      expect(good.status()).toBe(200);

      const write = await authorized.post(`${PUBLIC_BASE_URL}/items`, {
        headers: { 'content-type': 'application/json', 'x-tidal-ack': 'quorum' },
        data: {
          entity_id: 999_000_099,
          metadata: { title: 'deploy verification probe', category: 'verification' },
        },
      });
      expect(write.status()).toBe(201);

      records.push(
        await captureProofPanel(
          page,
          testInfo,
          {
            captureId: 'CAP-006-quorum-write',
            capabilityId: 'CAP-006',
            title: 'The boundary holds, and a write is committed by quorum',
            subtitle:
              'One hostname on the open internet. The same path, three credentials — then a real ' +
              'write that a majority of nodes acknowledged.',
            blocks: [
              {
                command: `curl -s -o /dev/null -w '%{http_code}' ${PUBLIC_HOST}/search   # no credential`,
                output: String(none.status()),
                verdict: 'Refused. Expected — the corpus is not public.',
                negative: true,
              },
              {
                command: `curl -H 'Authorization: Bearer <wrong>' ${PUBLIC_HOST}/search`,
                output: String(bad.status()),
                verdict: 'Refused. A wrong key is rejected in constant time.',
                negative: true,
              },
              {
                command: `curl -H 'Authorization: Bearer <key>' ${PUBLIC_HOST}/search`,
                output: String(good.status()),
                verdict: 'Served.',
              },
              {
                command: `curl -X POST -H 'x-tidal-ack: quorum' ${PUBLIC_HOST}/items`,
                output: `${write.status()} Created`,
                verdict:
                  '201 means a QUORUM acknowledged the write — not that one node accepted it. ' +
                  'DNS, TLS, gateway, auth and Raft replication, proven in one request.',
              },
            ],
            footnote:
              'Verification writes use entity ids in a reserved 999_000_0xx band so a probe is ' +
              'never mistaken for corpus data.',
          },
          {
            expected: '401, 401, 200, then 201 for a quorum-acked write',
            businessPurpose:
              'The single strongest available proof: the full stack works and the data plane is closed to strangers',
            personas: OPERATOR,
          },
        ),
      );
    } finally {
      await anonymous.dispose();
      await wrong.dispose();
      await authorized.dispose();
    }
  });

  test('CAP-008 the metrics port is closed to foreign pods but open to the scraper', async ({
    page,
  }, testInfo) => {
    const ip = await kubectl([
      '-n',
      NAMESPACE,
      'get',
      'pod',
      'tidaldb-0',
      '-o',
      'jsonpath={.status.podIP}',
    ]);
    const target = ip.stdout.trim();

    const foreign = await kubectl([
      '-n',
      FOREIGN_NAMESPACE,
      'exec',
      FOREIGN_POD,
      '--',
      'sh',
      '-c',
      `wget -qO- --timeout=5 http://${target}:${PORT_METRICS}/metrics 2>&1 | head -2`,
    ]);
    const scraper = await kubectl(
      [
        '-n',
        OBS_NAMESPACE,
        'exec',
        'deploy/vmagent',
        '--',
        'sh',
        '-c',
        `wget -qO- --timeout=15 http://${target}:${PORT_METRICS}/metrics | grep -c '^tidaldb_'`,
      ],
      { timeoutMs: 60_000 },
    );

    const foreignOutput = `${foreign.stdout}${foreign.stderr}`.trim();
    expect(foreignOutput).not.toContain('tidaldb_');
    const series = Number.parseInt(scraper.stdout.trim(), 10);
    expect(series).toBeGreaterThan(100);

    records.push(
      await captureProofPanel(
        page,
        testInfo,
        {
          captureId: 'CAP-008-network-isolation',
          capabilityId: 'CAP-008',
          title: 'The unauthenticated metrics port is not cluster-wide',
          subtitle:
            'A NetworkPolicy that blocks everything is an observability outage; one that blocks ' +
            'nothing is theatre. Both directions are shown.',
          blocks: [
            {
              command: `kubectl -n ${FOREIGN_NAMESPACE} exec ${FOREIGN_POD} -- wget http://${target}:9091/metrics`,
              output: foreignOutput.slice(0, 220),
              verdict:
                'Refused. Before the policy existed, any pod in the cluster could read the corpus size.',
              negative: true,
            },
            {
              command: `kubectl -n ${OBS_NAMESPACE} exec deploy/vmagent -- wget … | grep -c '^tidaldb_'`,
              output: `${series}`,
              verdict: `The scraper still collects ${series} series — monitoring is intact.`,
            },
          ],
          footnote:
            'Port 9500 is deliberately left open: all three probes originate from the node, and a ' +
            'wrong rule there restarts every pod.',
        },
        {
          expected: 'Connection refused from a foreign namespace; hundreds of series to the scraper',
          businessPurpose: 'Least-privilege network access without blinding the monitoring stack',
          personas: OPERATOR,
        },
      ),
    );
  });

  test('CAP-010 the operator dashboard renders live data', async ({ page }, testInfo) => {
    await withPortForward(OBS_NAMESPACE, 'deploy/grafana', 3000, async (forward) => {
      await page.setViewportSize({ width: CAPTURE_WIDTH, height: 1800 });
      await page.setExtraHTTPHeaders({
        authorization: `Basic ${Buffer.from(`admin:${grafanaPassword()}`).toString('base64')}`,
      });
      await page.goto(
        `http://127.0.0.1:${forward.localPort}/d/${DASHBOARD_UID}/?from=now-6h&to=now&refresh=`,
        { waitUntil: 'networkidle' },
      );
      await page
        .waitForFunction(() => document.querySelectorAll('canvas, .uplot').length > 3, {
          timeout: 90_000,
        })
        .catch(() => undefined);
      await page.waitForTimeout(6_000);

      const health = await page.locator('[data-panelid="14"]').innerText();
      expect(health, 'the health stat must render a value, not an empty box').toMatch(/OK|DOWN/);

      // Clip to the evidence band rather than shipping the whole 1600x1800
      // board. Scaled into a 16:9 frame the full board became illegible — the
      // audit protocol's rule is to crop to the relevant region, not to shrink
      // a dense page until nobody can read it. Bounds come from the real
      // elements so a layout change cannot silently mis-crop. See BUG-010.
      // Anchor the top to the ROW header (panel 7) so its title is not sliced,
      // and the bottom tight to the last stat panel so the next row does not
      // bleed in as a sliver. Both were visible crop artefacts on the first
      // attempt.
      const rowHeader = await page.locator('[data-panelid="7"]').boundingBox();
      const last = await page.locator('[data-panelid="19"]').boundingBox();
      expect(rowHeader, 'latency row header must anchor the top of the crop').not.toBeNull();
      expect(last, 'indexed vectors panel must anchor the bottom of the crop').not.toBeNull();

      const top = Math.max(0, Math.floor(rowHeader!.y - 8));
      const bottom = Math.ceil(last!.y + last!.height + 8);
      const clip = {
        x: 0,
        y: top,
        width: CAPTURE_WIDTH,
        height: bottom - top,
      };

      records.push(
        await recordScreenshot(await page.screenshot({ clip }), testInfo, {
          captureId: 'CAP-010-dashboard',
          capabilityId: 'CAP-010',
          expected:
            'Cluster health OK, reseed none, corpus size, and per-node latency charts — legible at delivery resolution',
          businessPurpose:
            'The first surface an operator opens during an incident actually shows the cluster',
          personas: OPERATOR,
          width: clip.width,
          height: clip.height,
        }),
      );
    });
  });

  test('CAP-014 operator authority is separate from data access', async ({
    page,
    playwright,
  }, testInfo) => {
    const admin = adminKey();
    expect(admin, 'admin key must be present').toBeTruthy();

    await withPortForward(NAMESPACE, 'svc/tidaldb', PORT_CLIENT, async (forward) => {
      const base = `https://127.0.0.1:${forward.localPort}`;
      const body = { region: 'tidaldb-1' };

      const dataContext = await playwright.request.newContext({
        ignoreHTTPSErrors: true,
        extraHTTPHeaders: {
          authorization: `Bearer ${apiKey()}`,
          'content-type': 'application/json',
        },
      });
      const adminContext = await playwright.request.newContext({
        ignoreHTTPSErrors: true,
        extraHTTPHeaders: {
          authorization: `Bearer ${admin}`,
          'content-type': 'application/json',
        },
      });

      try {
        const dataAttempt = await dataContext.post(`${base}/cluster/promote`, { data: body });
        const adminAttempt = await adminContext.post(`${base}/cluster/promote`, { data: body });
        expect(dataAttempt.status()).toBe(403);
        expect(adminAttempt.status()).not.toBe(403);

        records.push(
          await captureProofPanel(
            page,
            testInfo,
            {
              captureId: 'CAP-014-authority',
              capabilityId: 'CAP-014',
              title: 'An application key cannot remove a cluster member',
              subtitle:
                'The data credential authenticates but is not authorised for destructive ' +
                'operator verbs. Only the admin key is.',
              blocks: [
                {
                  command: "POST /cluster/promote   Authorization: Bearer <data key>",
                  output: `${dataAttempt.status()} Forbidden`,
                  verdict:
                    '403, not 401 — the key is valid, it simply lacks operator authority.',
                  negative: true,
                },
                {
                  command: 'POST /cluster/promote   Authorization: Bearer <admin key>',
                  output: `${adminAttempt.status()}`,
                  verdict: 'Authorised. Operator authority is a separate credential.',
                },
              ],
              footnote:
                'Before this split, the key every client holds could remove a member, force a ' +
                'partition, or transfer a shard.',
            },
            {
              expected: '403 for the data credential, not-403 for the admin credential',
              businessPurpose:
                'Blast radius of a leaked application key is bounded to data, not cluster topology',
              personas: OPERATOR,
            },
          ),
        );
      } finally {
        await dataContext.dispose();
        await adminContext.dispose();
      }
    });
  });

  test('CAP-014 CAP-015 the harness corrected its own runbook', async ({
    page,
    playwright,
  }, testInfo) => {
    // The dream beat needs a capture that SHOWS the contradiction, not one that
    // merely sits next to a caption describing it. Left: the claim exactly as it
    // was committed. Right: the live probe that disproved it. Both are real —
    // the doc text comes out of git, the status codes out of the cluster.
    const staleClaim = await run('git', [
      'show',
      'd21a202:docs/runbooks/deploy-verification.md',
    ]);
    expect(staleClaim.code, 'the original runbook revision must be readable').toBe(0);
    const claimLines = staleClaim.stdout
      .split('\n')
      .slice(383, 388)
      .join('\n')
      .trimEnd();
    expect(claimLines, 'expected the superseded section heading').toContain(
      'Not active yet',
    );

    const admin = adminKey();
    expect(admin, 'admin key must be present').toBeTruthy();

    const probe = await withPortForward(
      NAMESPACE,
      'svc/tidaldb',
      PORT_CLIENT,
      async (forward) => {
        const base = `https://127.0.0.1:${forward.localPort}`;
        const body = { region: 'tidaldb-1' };
        const dataContext = await playwright.request.newContext({
          ignoreHTTPSErrors: true,
          extraHTTPHeaders: {
            authorization: `Bearer ${apiKey()}`,
            'content-type': 'application/json',
          },
        });
        const adminContext = await playwright.request.newContext({
          ignoreHTTPSErrors: true,
          extraHTTPHeaders: {
            authorization: `Bearer ${admin}`,
            'content-type': 'application/json',
          },
        });
        try {
          const dataStatus = (await dataContext.post(`${base}/cluster/promote`, { data: body })).status();
          const adminStatus = (await adminContext.post(`${base}/cluster/promote`, { data: body })).status();
          return { dataStatus, adminStatus };
        } finally {
          await dataContext.dispose();
          await adminContext.dispose();
        }
      },
    );

    expect(probe.dataStatus, 'the gate must be enforcing').toBe(403);
    expect(probe.adminStatus, 'the admin key must be authorised').not.toBe(403);

    const image = await kubectl([
      '-n',
      NAMESPACE,
      'get',
      'statefulset',
      'tidaldb',
      '-o',
      'jsonpath={.spec.template.spec.containers[0].image}',
    ]);

    records.push(
      await captureProofPanel(
        page,
        testInfo,
        {
          captureId: 'CAP-014-drift',
          capabilityId: 'CAP-014',
          title: 'The document was wrong, and the harness said so',
          subtitle:
            'The runbook described a security control as not yet active. The harness read the ' +
            'live image and the live secret, and found it already enforcing.',
          blocks: [
            {
              command: 'git show d21a202:docs/runbooks/deploy-verification.md   # as committed',
              output: claimLines,
              verdict: 'The claim: inert, pending an image roll.',
              negative: true,
            },
            {
              command: 'kubectl get statefulset tidaldb -o jsonpath={..image}',
              output: image.stdout.trim(),
              verdict: 'A different image is running than the one the runbook described.',
            },
            {
              command: 'POST /cluster/promote  with the data key, then the admin key',
              output: `data key  -> ${probe.dataStatus} Forbidden\nadmin key -> ${probe.adminStatus}`,
              verdict:
                'The gate was live the whole time. Documentation drift, caught by the thing it documents.',
            },
          ],
          footnote:
            'Recorded as BUG-001. Section 9 of the runbook was rewritten in the same pass that ' +
            'found this.',
        },
        {
          expected: 'The superseded claim beside the live probe that contradicts it',
          businessPurpose:
            'Verification that audits its own documentation instead of drifting away from it',
          personas: OPERATOR,
        },
      ),
    );
  });

  test('CAP-013 the fleet backup captured every volume', async ({ page }, testInfo) => {
    const list = await kubectl([
      '-n',
      BACKUP_NAMESPACE,
      'get',
      'backup.velero.io',
      '-l',
      `velero.io/schedule-name=${BACKUP_SCHEDULE}`,
      '--sort-by=.metadata.creationTimestamp',
      '-o',
      'jsonpath={range .items[*]}{.metadata.name}{"\\n"}{end}',
    ]);
    const names = list.stdout.trim().split('\n').filter(Boolean);
    const newest = names[names.length - 1];
    expect(newest).not.toMatch(/restore-canary/);

    const detail = await kubectl([
      '-n',
      BACKUP_NAMESPACE,
      'get',
      'backup.velero.io',
      newest,
      '-o',
      'jsonpath=phase={.status.phase} items={.status.progress.itemsBackedUp}/{.status.progress.totalItems}',
    ]);
    const volumes = await kubectl([
      '-n',
      BACKUP_NAMESPACE,
      'get',
      'podvolumebackups',
      '-l',
      `velero.io/backup-name=${newest}`,
      '-o',
      'jsonpath={range .items[*]}{.status.phase}{"\\n"}{end}',
    ]);
    const phases = volumes.stdout.trim().split('\n').filter(Boolean);
    expect(detail.stdout).toContain('phase=Completed');
    expect([...new Set(phases)]).toEqual(['Completed']);

    records.push(
      await captureProofPanel(
        page,
        testInfo,
        {
          captureId: 'CAP-013-backup',
          capabilityId: 'CAP-013',
          title: 'The recovery story is intact',
          subtitle:
            'Selected by the schedule label the freshness alert actually watches — not simply the ' +
            'newest backup object.',
          blocks: [
            {
              command: `kubectl -n ${BACKUP_NAMESPACE} get backup.velero.io -l velero.io/schedule-name=${BACKUP_SCHEDULE} | tail -1`,
              output: `${newest}\n${detail.stdout.trim()}`,
              verdict: 'Completed with every discovered item captured.',
            },
            {
              command: `kubectl -n ${BACKUP_NAMESPACE} get podvolumebackups -l velero.io/backup-name=${newest}`,
              output: `${phases.length} PodVolumeBackups, all ${[...new Set(phases)].join(', ')}`,
              verdict:
                'One failed volume marks the whole backup PartiallyFailed and freezes the ' +
                'freshness alert — so every volume must be clean.',
            },
          ],
          footnote:
            'Sorting all backups by timestamp instead would have selected a restore-canary run: ' +
            '20 items, one volume, and a meaningless pass.',
        },
        {
          expected: 'Completed, all items, every PodVolumeBackup Completed',
          businessPurpose: 'The cluster can actually be restored, and the alert is trustworthy',
          personas: OPERATOR,
        },
      ),
    );
  });

  test('CAP-015 what is NOT verified is stated', async ({ page }, testInfo) => {
    const ip = await kubectl([
      '-n',
      NAMESPACE,
      'get',
      'pod',
      'tidaldb-0',
      '-o',
      'jsonpath={.status.podIP}',
    ]);
    const counts = await kubectl(
      [
        '-n',
        OBS_NAMESPACE,
        'exec',
        'deploy/vmagent',
        '--',
        'sh',
        '-c',
        `wget -qO- --timeout=15 http://${ip.stdout.trim()}:${PORT_METRICS}/metrics ` +
          `| awk '/^tidaldb_http_/{h++} /^tidaldb_/{t++} END{print "tidaldb_* = "(t+0)"\\ntidaldb_http_* = "(h+0)}'`,
      ],
      { timeoutMs: 60_000 },
    );
    const image = await kubectl([
      '-n',
      NAMESPACE,
      'get',
      'statefulset',
      'tidaldb',
      '-o',
      'jsonpath={.spec.template.spec.containers[0].image}',
    ]);

    expect(counts.stdout).toMatch(/tidaldb_http_\* = 0/);

    records.push(
      await captureProofPanel(
        page,
        testInfo,
        {
          captureId: 'CAP-015-inert',
          capabilityId: 'CAP-015',
          title: 'What this deployment does not yet do',
          subtitle:
            'Two committed features are absent from the running image. Stating that is part of ' +
            'the verification, not a footnote to it.',
          blocks: [
            {
              command: 'kubectl get statefulset tidaldb -o jsonpath={..image}',
              output: image.stdout.trim(),
              verdict: 'This image carries the credential split but predates the observability commit.',
            },
            {
              command: 'scrape :9091 and count metric families',
              output: counts.stdout.trim(),
              verdict:
                'Zero HTTP metrics — and the baseline count proves the scrape WORKED, so "absent" ' +
                'is distinguishable from "unscraped". Five dashboard panels are legitimately empty.',
              negative: true,
            },
          ],
          footnote:
            'The suite asserts this absence deliberately: the day the observability image is ' +
            'rolled, these tests fail and say so — instead of the runbook silently rotting.',
        },
        {
          expected: 'Zero tidaldb_http_* families while baseline tidaldb_* families are present',
          businessPurpose:
            'A verification that hides its gaps cannot be trusted about the parts it claims',
          personas: OPERATOR,
        },
      ),
    );
  });

  test('CAP-012 tidalctl gives an operator a live view and an exit code', async ({
    page,
  }, testInfo) => {
    await withPortForward(NAMESPACE, 'svc/tidaldb', PORT_CLIENT, async (forward) => {
      const result = await tidalctl(
        [
          'cluster-status',
          '--url',
          `https://127.0.0.1:${forward.localPort}`,
          '--key',
          apiKey(),
          '--insecure',
        ],
        { timeoutMs: 45_000 },
      );
      expect(result.stdout).toContain('regions:');

      records.push(
        await captureProofPanel(
          page,
          testInfo,
          {
            captureId: 'CAP-012-tidalctl',
            capabilityId: 'CAP-012',
            title: 'One command, and it names its own blind spot',
            subtitle:
              'The aggregated endpoint reports peers it holds no frontier report for as zero. ' +
              'tidalctl labels that instead of repeating it as lag.',
            blocks: [
              {
                command: 'tidalctl cluster-status --url https://… --insecure',
                output: result.stdout.trim(),
                // negative: the exit code is the FINDING, not a success. Rendering
                // it green would have colour implying "good" for a defect.
                negative: true,
                verdict:
                  'NO REPORT is an honest "I do not know", not a fabricated 13.3M-event deficit. ' +
                  `But exit code ${result.code} on a converged cluster makes the documented ` +
                  '`cluster-status && deploy` gate unusable.',
              },
            ],
            footnote:
              'Consequence, recorded as BUG-005: because a converged cluster still exits 2, ' +
              '`tidalctl cluster-status && deploy` is NOT a usable gate on this deployment.',
          },
          {
            expected: 'Leader, region table with NO REPORT markers, shard table, exit 2',
            businessPurpose:
              'An operator can interrogate the cluster without hand-rolling curl, and is told what the tool cannot see',
            personas: OPERATOR,
          },
        ),
      );
    });
  });
});