from __future__ import annotations

import hashlib
import json
import os
import socket
import subprocess
import sys
import threading
from contextlib import contextmanager
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from typing import Any, Iterator


ROOT = Path(__file__).resolve().parents[3]
EXPORTER = ROOT / "scripts" / "backup" / "signoz-metadata-export.py"
VERIFIER = ROOT / "scripts" / "backup" / "verify-signoz-metadata-export.py"
RESTORE = ROOT / "scripts" / "backup" / "signoz-metadata-restore-drill.py"
POLICY = ROOT / "config" / "signoz" / "metadata-export-policy.json"
TRACE_ID = "trace-P0-OBS-002-metadata"
RUN_ID = "run-P0-OBS-002-metadata"
WORK_ITEM_ID = "P0-OBS-002"
TOKEN = "test-only-signoz-token-reference"


SOURCE_ASSETS: dict[str, Any] = {
    "/api/v1/dashboards": {"data": [{"id": "dash-1", "title": "Ops"}]},
    "/api/v1/rules": {"data": [{"id": "rule-1", "name": "Latency"}]},
    "/api/v1/explorer/views": {"data": [{"id": "view-1", "name": "Errors"}]},
    "/api/v1/roles": {"data": [{"id": "role-1", "name": "Viewer"}]},
    "/api/v1/org/preferences": {"data": [{"name": "theme", "value": "dark"}]},
    "/api/v1/user/preferences": {"data": [{"name": "timezone", "value": "UTC"}]},
    "/api/v1/global/config": {"data": {"setupCompleted": True}},
}


class MetadataHandler(BaseHTTPRequestHandler):
    assets: dict[str, Any] = {}
    drift_path: str | None = None
    get_counts: dict[str, int] = {}
    post_count = 0

    def log_message(self, format: str, *args: object) -> None:
        return

    def _authorized(self) -> bool:
        return self.headers.get("SIGNOZ-API-KEY") == TOKEN

    def _send_json(self, status: int, value: Any) -> None:
        payload = json.dumps(value, separators=(",", ":")).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(payload)))
        self.end_headers()
        self.wfile.write(payload)

    def do_GET(self) -> None:
        if not self._authorized():
            self._send_json(401, {"error": "unauthorized"})
            return
        path = self.path.split("?", 1)[0]
        if path not in self.assets:
            self._send_json(404, {"error": "not found"})
            return
        self.get_counts[path] = self.get_counts.get(path, 0) + 1
        value = json.loads(json.dumps(self.assets[path]))
        if self.drift_path == path and self.get_counts[path] >= 2:
            value["data"].append({"id": "drift", "title": "changed"})
        self._send_json(200, value)

    def do_POST(self) -> None:
        if not self._authorized():
            self._send_json(401, {"error": "unauthorized"})
            return
        path = self.path.split("?", 1)[0]
        if path not in self.assets or not isinstance(
            self.assets[path].get("data"), list
        ):
            self._send_json(404, {"error": "not found"})
            return
        length = int(self.headers.get("Content-Length", "0"))
        value = json.loads(self.rfile.read(length))
        if not isinstance(value, dict):
            self._send_json(400, {"error": "object required"})
            return
        type(self).post_count += 1
        restored = {"id": f"server-{type(self).post_count}", **value}
        self.assets[path]["data"].append(restored)
        self._send_json(201, {"data": restored})


@contextmanager
def metadata_server(
    assets: dict[str, Any],
    *,
    drift_path: str | None = None,
) -> Iterator[tuple[str, type[MetadataHandler]]]:
    handler = type("BoundMetadataHandler", (MetadataHandler,), {})
    handler.assets = json.loads(json.dumps(assets))
    handler.drift_path = drift_path
    handler.get_counts = {}
    handler.post_count = 0
    server = ThreadingHTTPServer(("127.0.0.1", 0), handler)
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    try:
        host, port = server.server_address
        yield f"http://{host}:{port}", handler
    finally:
        server.shutdown()
        server.server_close()
        thread.join(timeout=5)


def credential_file(tmp_path: Path, *, mode: int = 0o600) -> Path:
    path = tmp_path / "signoz-api-key.ref"
    path.write_text(f"{TOKEN}\n", encoding="ascii")
    path.chmod(mode)
    return path


def export_command(
    *,
    base_url: str,
    output_dir: Path,
    credential: Path | None,
    apply: bool,
    receipt_file: Path | None = None,
) -> list[str]:
    command = [
        sys.executable,
        str(EXPORTER),
        "--apply" if apply else "--check",
        "--base-url",
        base_url,
        "--output-dir",
        str(output_dir),
        "--trace-id",
        TRACE_ID,
        "--run-id",
        RUN_ID,
        "--work-item-id",
        WORK_ITEM_ID,
    ]
    if credential is not None:
        command.extend(["--credential-file", str(credential)])
    if apply:
        terminal_path = receipt_file or output_dir.with_name(
            f"{output_dir.name}.terminal.json"
        )
        command.extend(["--receipt-file", str(terminal_path)])
    return command


def create_bundle(tmp_path: Path) -> tuple[Path, Path]:
    credential = credential_file(tmp_path)
    bundle = tmp_path / "bundle"
    with metadata_server(SOURCE_ASSETS) as (base_url, _):
        result = subprocess.run(
            export_command(
                base_url=base_url,
                output_dir=bundle,
                credential=credential,
                apply=True,
            ),
            text=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            timeout=20,
            check=False,
        )
    assert result.returncode == 0, result.stderr
    return bundle, credential


def isolation_receipt(tmp_path: Path, target_base_url: str) -> Path:
    policy = json.loads(POLICY.read_text(encoding="utf-8"))
    value = {
        "schema": "awoooi_signoz_metadata_restore_isolation_v1",
        "trace_id": TRACE_ID,
        "run_id": RUN_ID,
        "work_item_id": WORK_ITEM_ID,
        "target": {
            "canonical_id": f"ephemeral-signoz-metadata-restore-{RUN_ID}",
            "base_url_sha256": hashlib.sha256(
                target_base_url.encode("utf-8")
            ).hexdigest(),
            "image_ref": "signoz/signoz:v0.113.0",
            "image_id": f"sha256:{'a' * 64}",
            "image_present_locally": True,
            "image_pull_performed": False,
            "production_network_attached": False,
            "production_volume_attached": False,
            "ephemeral_volumes_only": True,
            "cleanup_controller_armed": True,
            "zero_residue_verifier_armed": True,
            "network_scope": "loopback_and_internal_only",
            "resource_label_key": policy["restore_isolation"]["resource_label_key"],
            "resource_label_value": RUN_ID,
        },
    }
    path = tmp_path / "isolation.json"
    path.write_text(json.dumps(value), encoding="utf-8")
    path.chmod(0o600)
    return path


def restore_command(
    *,
    mode: str,
    bundle: Path,
    target_base_url: str,
    credential: Path | None,
    isolation: Path,
    receipt_file: Path | None = None,
) -> list[str]:
    command = [
        sys.executable,
        str(RESTORE),
        mode,
        "--bundle-dir",
        str(bundle),
        "--target-base-url",
        target_base_url,
        "--isolation-receipt",
        str(isolation),
        "--trace-id",
        TRACE_ID,
        "--run-id",
        RUN_ID,
        "--work-item-id",
        WORK_ITEM_ID,
    ]
    if credential is not None:
        command.extend(["--credential-file", str(credential)])
    if mode == "--apply":
        terminal_path = receipt_file or bundle.parent / "restore-terminal.json"
        command.extend(["--receipt-file", str(terminal_path)])
    return command


def test_policy_is_explicitly_non_raw_and_non_completion() -> None:
    policy = json.loads(POLICY.read_text(encoding="utf-8"))
    assert policy["source_runtime"]["raw_database_access_allowed"] is False
    assert policy["source_runtime"]["raw_volume_access_allowed"] is False
    assert policy["source_runtime"]["github_supply_chain_allowed"] is False
    assert policy["completion_contract"]["portable_export_alone_is_completion"] is False
    assert (
        policy["completion_contract"]["full_sqlite_completion_claim_allowed"] is False
    )
    assert {item["path"] for item in policy["forbidden_endpoints"]} >= {
        "/api/v1/pats",
        "/api/v1/domains",
        "/api/v1/channels",
    }


def test_check_mode_writes_nothing(tmp_path: Path) -> None:
    output = tmp_path / "must-not-exist"
    result = subprocess.run(
        export_command(
            base_url="https://signoz.invalid",
            output_dir=output,
            credential=None,
            apply=False,
        ),
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        timeout=10,
        check=False,
    )
    assert result.returncode == 0, result.stderr
    assert json.loads(result.stdout)["terminal"] == "check_pass_no_write"
    assert not output.exists()


def test_apply_and_independent_verifier_pass(tmp_path: Path) -> None:
    bundle, _ = create_bundle(tmp_path)
    result = subprocess.run(
        [
            sys.executable,
            str(VERIFIER),
            "--bundle-dir",
            str(bundle),
            "--expected-trace-id",
            TRACE_ID,
            "--expected-run-id",
            RUN_ID,
            "--expected-work-item-id",
            WORK_ITEM_ID,
        ],
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        timeout=10,
        check=False,
    )
    assert result.returncode == 0, result.stderr
    assert json.loads(result.stdout)["terminal"] == (
        "pass_export_verified_restore_pending"
    )
    manifest = json.loads((bundle / "manifest.json").read_text(encoding="utf-8"))
    assert manifest["stable_read_count"] == 2
    assert manifest["raw_sqlite_read"] is False
    assert manifest["sensitive_value_persisted"] is False
    assert manifest["completion_claim"] is False
    assert oct((bundle / "manifest.json").stat().st_mode & 0o777) == "0o600"


def test_secret_key_fails_closed_without_bundle(tmp_path: Path) -> None:
    assets = json.loads(json.dumps(SOURCE_ASSETS))
    assets["/api/v1/global/config"]["data"]["clientSecret"] = "must-not-persist"
    output = tmp_path / "bundle"
    credential = credential_file(tmp_path)
    with metadata_server(assets) as (base_url, _):
        result = subprocess.run(
            export_command(
                base_url=base_url,
                output_dir=output,
                credential=credential,
                apply=True,
            ),
            text=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            timeout=20,
            check=False,
        )
    assert result.returncode == 1
    assert "forbidden_key_detected" in result.stderr
    assert "must-not-persist" not in result.stderr
    assert not output.exists()


def test_source_drift_fails_closed_without_bundle(tmp_path: Path) -> None:
    output = tmp_path / "bundle"
    credential = credential_file(tmp_path)
    with metadata_server(SOURCE_ASSETS, drift_path="/api/v1/dashboards") as (
        base_url,
        _,
    ):
        result = subprocess.run(
            export_command(
                base_url=base_url,
                output_dir=output,
                credential=credential,
                apply=True,
            ),
            text=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            timeout=20,
            check=False,
        )
    assert result.returncode == 1
    assert "source_metadata_drift_between_stable_reads" in result.stderr
    assert not output.exists()


def test_tamper_breaks_independent_verifier(tmp_path: Path) -> None:
    bundle, _ = create_bundle(tmp_path)
    asset = bundle / "assets" / "dashboards.json"
    asset.write_text('{"data":[]}\n', encoding="utf-8")
    asset.chmod(0o600)
    result = subprocess.run(
        [
            sys.executable,
            str(VERIFIER),
            "--bundle-dir",
            str(bundle),
            "--expected-trace-id",
            TRACE_ID,
            "--expected-run-id",
            RUN_ID,
        ],
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        timeout=10,
        check=False,
    )
    assert result.returncode == 1
    assert "manifest_asset_hash_mismatch_dashboards" in result.stderr


def test_unexpected_bundle_file_breaks_independent_verifier(tmp_path: Path) -> None:
    bundle, _ = create_bundle(tmp_path)
    unexpected = bundle / "unlisted.json"
    unexpected.write_text('{"not":"manifested"}\n', encoding="utf-8")
    unexpected.chmod(0o600)
    result = subprocess.run(
        [
            sys.executable,
            str(VERIFIER),
            "--bundle-dir",
            str(bundle),
            "--expected-trace-id",
            TRACE_ID,
            "--expected-run-id",
            RUN_ID,
        ],
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        timeout=10,
        check=False,
    )
    assert result.returncode == 1
    assert "bundle_tree_contains_unexpected_entries" in result.stderr


def test_restore_check_and_semantic_apply_are_bounded(tmp_path: Path) -> None:
    bundle, credential = create_bundle(tmp_path)
    target_assets = {
        "/api/v1/dashboards": {"data": []},
        "/api/v1/rules": {"data": []},
        "/api/v1/explorer/views": {"data": []},
    }
    with metadata_server(target_assets) as (target_base_url, handler):
        isolation = isolation_receipt(tmp_path, target_base_url)
        check_result = subprocess.run(
            restore_command(
                mode="--check",
                bundle=bundle,
                target_base_url=target_base_url,
                credential=credential,
                isolation=isolation,
            ),
            text=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            timeout=20,
            check=False,
        )
        assert check_result.returncode == 0, check_result.stderr
        assert json.loads(check_result.stdout)["terminal"] == "check_pass_no_write"
        assert handler.post_count == 0

        apply_result = subprocess.run(
            restore_command(
                mode="--apply",
                bundle=bundle,
                target_base_url=target_base_url,
                credential=credential,
                isolation=isolation,
            ),
            text=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            timeout=20,
            check=False,
        )
        assert apply_result.returncode == 0, apply_result.stderr
        assert json.loads(apply_result.stdout)["terminal"] == (
            "partial_degraded_cleanup_pending"
        )
        phase_receipts = json.loads(apply_result.stdout)["phase_receipts"]
        assert [item["phase"] for item in phase_receipts] == [
            "sensor_source",
            "normalized_asset_identity",
            "source_of_truth_diff",
            "ai_decision",
            "risk_policy_decision",
            "check_mode",
            "bounded_execution",
            "independent_post_verifier",
            "rollback",
            "learning_writeback",
        ]
        assert phase_receipts[-1]["terminal"] == "pending"
        assert handler.post_count == 3


def test_restore_rejects_non_loopback_target_before_writes(tmp_path: Path) -> None:
    bundle, credential = create_bundle(tmp_path)
    target = "https://signoz.example.invalid"
    isolation = isolation_receipt(tmp_path, target)
    result = subprocess.run(
        restore_command(
            mode="--apply",
            bundle=bundle,
            target_base_url=target,
            credential=credential,
            isolation=isolation,
        ),
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        timeout=20,
        check=False,
    )
    assert result.returncode == 1
    assert "restore_target_must_be_loopback" in result.stderr


def test_cleanup_verifier_requires_zero_labeled_residue(tmp_path: Path) -> None:
    bundle, _ = create_bundle(tmp_path)
    fake_bin = tmp_path / "bin"
    fake_bin.mkdir()
    docker = fake_bin / "docker"
    docker.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8")
    docker.chmod(0o755)

    with socket.socket() as sock:
        sock.bind(("127.0.0.1", 0))
        port = sock.getsockname()[1]
    target = f"http://127.0.0.1:{port}"
    isolation = isolation_receipt(tmp_path, target)
    env = {**os.environ, "PATH": f"{fake_bin}:{os.environ['PATH']}"}
    result = subprocess.run(
        restore_command(
            mode="--verify-cleanup",
            bundle=bundle,
            target_base_url=target,
            credential=None,
            isolation=isolation,
        ),
        env=env,
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        timeout=20,
        check=False,
    )
    assert result.returncode == 0, result.stderr
    assert json.loads(result.stdout)["terminal"] == "pass_zero_residue_verified"


def test_credential_reference_mode_fails_before_network(tmp_path: Path) -> None:
    credential = credential_file(tmp_path, mode=0o644)
    output = tmp_path / "bundle"
    result = subprocess.run(
        export_command(
            base_url="https://signoz.invalid",
            output_dir=output,
            credential=credential,
            apply=True,
        ),
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        timeout=10,
        check=False,
    )
    assert result.returncode == 1
    assert "credential_reference_mode_must_be_0400_or_0600" in result.stderr
    assert not output.exists()


def test_credential_symlink_fails_before_network(tmp_path: Path) -> None:
    credential = credential_file(tmp_path)
    link = tmp_path / "credential-link"
    link.symlink_to(credential)
    output = tmp_path / "bundle"
    result = subprocess.run(
        export_command(
            base_url="https://signoz.invalid",
            output_dir=output,
            credential=link,
            apply=True,
        ),
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        timeout=10,
        check=False,
    )
    assert result.returncode == 1
    assert "credential_reference_symlink_component_forbidden" in result.stderr
    assert not output.exists()


def test_apply_failure_writes_terminal_receipt_without_secret(tmp_path: Path) -> None:
    assets = json.loads(json.dumps(SOURCE_ASSETS))
    secret_value = "must-not-appear-in-terminal-receipt"
    assets["/api/v1/global/config"]["data"]["clientSecret"] = secret_value
    output = tmp_path / "bundle"
    receipt_file = tmp_path / "terminal.json"
    credential = credential_file(tmp_path)
    with metadata_server(assets) as (base_url, _):
        command = export_command(
            base_url=base_url,
            output_dir=output,
            credential=credential,
            apply=True,
            receipt_file=receipt_file,
        )
        result = subprocess.run(
            command,
            text=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            timeout=20,
            check=False,
        )
    assert result.returncode == 1
    terminal = json.loads(receipt_file.read_text(encoding="utf-8"))
    assert terminal["terminal"] == "failed_export_not_closable"
    assert secret_value not in receipt_file.read_text(encoding="utf-8")
    assert not output.exists()