#!/usr/bin/env python3
"""Read-only, redacted inspection of an interrupted automatic-backup chain."""

from __future__ import annotations

import argparse
import hashlib
import json
import os
import re
import stat
import sys
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Iterable


BATCH_RE = re.compile(r"^hyb-oa-backup-(auto-(\d{8}T\d{6}Z))$")
PARTIAL_RE = re.compile(r"^hyb-oa-backup-(auto-(\d{8}T\d{6}Z))\.partial$")
TEMP_RE = re.compile(r"^\.hyb-oa-backup-.+\.tmp\..+$")
EXPECTED_FILES = {"database.dump", "storage.tar.gz", "manifest.json"}
ARTIFACTS = {"database": "database.dump", "storage": "storage.tar.gz"}
RSYNC_CODE_RE = re.compile(r"rsync error:[^\r\n]*\(code (11|12|23|30|255)\)")
RSYNC_CATEGORIES = {
    "11": "file_io_error",
    "12": "protocol_stream_error",
    "23": "partial_transfer_error",
    "30": "transport_timeout",
    "255": "ssh_session_error",
}
SIMILAR_RELATIVE_PPM = 50_000
SIMILAR_ABSOLUTE_BYTES = 64 * 1024 * 1024
SAFE_KEY_RE = re.compile(r"^[a-z0-9_]+$")
SAFE_VALUE_RE = re.compile(r"^(?:[A-Za-z][A-Za-z0-9_]*|-?[0-9]+)$")
SHA256_RE = re.compile(r"^[0-9a-f]{64}$")


class InspectionError(Exception):
    """Stable, non-sensitive inspection failure."""

    def __init__(self, category: str):
        super().__init__(category)
        self.category = category


@dataclass(frozen=True)
class BatchResult:
    valid: bool
    total_bytes: int
    created_epoch: int | None
    expected_artifact_bytes: int | None
    received_artifact_bytes: int


def emit(items: Iterable[tuple[str, object]]) -> None:
    rendered_items: list[tuple[str, str]] = []
    for key, value in items:
        if isinstance(value, bool):
            rendered = "yes" if value else "no"
        else:
            rendered = str(value)
        if not SAFE_KEY_RE.fullmatch(key) or not SAFE_VALUE_RE.fullmatch(rendered):
            print("probe_result=FAIL")
            print("probe_category=output_allowlist_violation")
            raise SystemExit(1)
        rendered_items.append((key, rendered))
    for key, rendered in rendered_items:
        print(f"{key}={rendered}")


def safe_root(path: Path) -> Path:
    try:
        if not path.is_absolute() or path.is_symlink() or not path.is_dir():
            raise InspectionError("root_invalid")
        list(path.iterdir())
    except PermissionError as error:
        raise InspectionError("access_denied") from error
    except OSError as error:
        raise InspectionError("root_unreadable") from error
    return path


def file_sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def tree_bytes(path: Path) -> int:
    total = 0
    try:
        for root, directories, files in os.walk(path, followlinks=False):
            directories[:] = [
                name for name in directories if not (Path(root) / name).is_symlink()
            ]
            for name in files:
                item = Path(root) / name
                if item.is_symlink():
                    continue
                total += item.stat().st_size
    except PermissionError as error:
        raise InspectionError("access_denied") from error
    except OSError as error:
        raise InspectionError("entry_unreadable") from error
    return total


def run_epoch(stamp: str) -> int | None:
    try:
        parsed = datetime.strptime(stamp, "%Y%m%dT%H%M%SZ").replace(tzinfo=timezone.utc)
    except ValueError:
        return None
    return int(parsed.timestamp())


def validate_batch(
    path: Path,
    expected_run_id: str,
    stamp: str,
    *,
    verify_hashes: bool = True,
) -> BatchResult:
    total = tree_bytes(path)
    received_artifact_bytes = 0
    expected_artifact_bytes: int | None = None
    created_epoch = run_epoch(stamp)
    if created_epoch is None:
        return BatchResult(False, total, None, None, 0)
    try:
        entries = list(path.iterdir())
        regular_names = {
            item.name
            for item in entries
            if not item.is_symlink() and stat.S_ISREG(item.stat().st_mode)
        }
        if len(entries) != 3 or regular_names != EXPECTED_FILES:
            return BatchResult(False, total, created_epoch, None, 0)
        manifest = json.loads((path / "manifest.json").read_text(encoding="utf-8"))
        if (
            manifest.get("schemaVersion") != 1
            or manifest.get("status") != "complete"
            or manifest.get("runId") != expected_run_id
            or not (manifest.get("buildSha") or manifest.get("sourceImageDigest"))
        ):
            return BatchResult(False, total, created_epoch, None, 0)
        artifacts = manifest.get("artifacts") or {}
        expected_artifact_bytes = 0
        for key, expected_name in ARTIFACTS.items():
            item = artifacts.get(key) or {}
            target = path / expected_name
            if item.get("file") != expected_name or isinstance(item.get("bytes"), bool):
                return BatchResult(False, total, created_epoch, None, 0)
            expected_size = item.get("bytes")
            if not isinstance(expected_size, int) or expected_size < 0:
                return BatchResult(False, total, created_epoch, None, 0)
            expected_sha = item.get("sha256")
            if not isinstance(expected_sha, str) or not SHA256_RE.fullmatch(expected_sha):
                return BatchResult(False, total, created_epoch, None, 0)
            expected_artifact_bytes += expected_size
            actual_size = target.stat().st_size
            received_artifact_bytes += actual_size
            if actual_size != expected_size or (
                verify_hashes and file_sha256(target) != expected_sha
            ):
                return BatchResult(
                    False,
                    total,
                    created_epoch,
                    expected_artifact_bytes,
                    received_artifact_bytes,
                )
        return BatchResult(
            True,
            total,
            created_epoch,
            expected_artifact_bytes,
            received_artifact_bytes,
        )
    except PermissionError as error:
        raise InspectionError("access_denied") from error
    except (AttributeError, json.JSONDecodeError, OSError, TypeError, ValueError):
        return BatchResult(False, total, created_epoch, expected_artifact_bytes, received_artifact_bytes)


def inspect_partial(path: Path, run_id: str, stamp: str) -> BatchResult:
    """Measure a partial without requiring the complete three-file contract."""
    total = tree_bytes(path)
    received = 0
    expected: int | None = None
    try:
        manifest_path = path / "manifest.json"
        if manifest_path.is_file() and not manifest_path.is_symlink():
            manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
            if manifest.get("runId") == run_id and manifest.get("schemaVersion") == 1:
                artifacts = manifest.get("artifacts") or {}
                sizes: list[int] = []
                for key, expected_name in ARTIFACTS.items():
                    item = artifacts.get(key) or {}
                    size = item.get("bytes")
                    if (
                        item.get("file") != expected_name
                        or isinstance(size, bool)
                        or not isinstance(size, int)
                        or size < 0
                    ):
                        sizes = []
                        break
                    sizes.append(size)
                    target = path / expected_name
                    if target.is_file() and not target.is_symlink():
                        received += target.stat().st_size
                if len(sizes) == 2:
                    expected = sum(sizes)
        validated = validate_batch(path, run_id, stamp)
        return BatchResult(validated.valid, total, run_epoch(stamp), expected, received)
    except PermissionError as error:
        raise InspectionError("access_denied") from error
    except (AttributeError, json.JSONDecodeError, OSError, TypeError, ValueError):
        return BatchResult(False, total, run_epoch(stamp), expected, received)


def filesystem_fields(prefix: str, root: Path) -> list[tuple[str, object]]:
    values = os.statvfs(root)
    total = values.f_blocks * values.f_frsize
    available = values.f_bavail * values.f_frsize
    used_percent = 0 if total == 0 else ((total - available) * 100) // total
    return [
        (f"{prefix}_disk_used_percent", used_percent),
        (f"{prefix}_disk_free_gib", available // (1024**3)),
    ]


def source_contract(wrapper: Path, backup_script: Path) -> list[tuple[str, object]]:
    try:
        if (
            wrapper.is_symlink()
            or not wrapper.is_file()
            or backup_script.is_symlink()
            or not backup_script.is_file()
        ):
            raise InspectionError("source_contract_invalid")
        wrapper_text = wrapper.read_text(encoding="utf-8")
        backup_text = backup_script.read_text(encoding="utf-8")
    except PermissionError as error:
        raise InspectionError("source_contract_access_denied") from error
    except OSError as error:
        raise InspectionError("source_contract_unreadable") from error

    if "run_backup() {" not in wrapper_text or "\ncheck_stale() {" not in wrapper_text:
        raise InspectionError("source_contract_invalid")
    run_backup_body = wrapper_text.split("run_backup() {", 1)[1].split("\ncheck_stale() {", 1)[0]
    markers = {
        "local_publish": '[[ -d "$batch_dir" && -f "$batch_dir/manifest.json" ]]',
        "transfer": "capture_command offhost_transfer rsync",
        "finalize": "capture_command offhost_finalize ssh",
        "success": 'write_success_state "$run_id" "$manifest_sha"',
        "prune": 'prune_local_backups "$batch_name"',
    }
    positions = {key: run_backup_body.find(marker) for key, marker in markers.items()}
    all_present = all(position >= 0 for position in positions.values())
    retained = all_present and positions["local_publish"] < positions["transfer"] < positions["prune"]
    pruned = all_present and positions["finalize"] < positions["success"] < positions["prune"]
    temp_cleanup = all(
        marker in backup_text
        for marker in (
            "trap cleanup EXIT",
            'TEMP_DIR="$(mktemp -d',
            'rm -rf -- "$TEMP_DIR"',
            'mv "$TEMP_DIR" "$FINAL_DIR"',
            'TEMP_DIR=""',
            "trap - EXIT",
        )
    )
    return [
        ("retained_after_transfer_failure", retained),
        ("pruned_after_offhost_success", pruned),
        ("generation_temp_cleanup_on_failure", temp_cleanup),
        ("cleanup_contract_source_verified", all_present and temp_cleanup),
    ]


def last_backup_phase(state_dir: Path) -> str:
    state_file = state_dir / "last-run.json"
    if not state_file.exists():
        return "NOT_AVAILABLE"
    try:
        if state_file.is_symlink() or not state_file.is_file():
            return "STATE_INVALID"
        payload = json.loads(state_file.read_text(encoding="utf-8"))
        findings = payload.get("findings") or []
        detail = findings[0].get("detail") if len(findings) == 1 and isinstance(findings[0], dict) else None
        match = re.fullmatch(r"phase_([a-z_]+)_failed", detail or "")
        allowed = {
            "runtime_identity",
            "paired_backup",
            "offhost_transfer",
            "offhost_finalize",
            "success_state",
            "stale_check",
            "quarterly_reminder",
        }
        return match.group(1) if match and match.group(1) in allowed else "NOT_AVAILABLE"
    except PermissionError as error:
        raise InspectionError("diagnostic_access_denied") from error
    except (AttributeError, json.JSONDecodeError, OSError, TypeError):
        return "STATE_INVALID"


def parse_transfer_diagnostic(state_dir: Path) -> list[tuple[str, object]]:
    phase = last_backup_phase(state_dir)
    diagnostic = state_dir / "last-error.log"
    if phase != "offhost_transfer" or not diagnostic.exists():
        code = "NOT_AVAILABLE"
    else:
        try:
            if diagnostic.is_symlink() or not diagnostic.is_file():
                raise InspectionError("diagnostic_invalid")
            text = diagnostic.read_text(encoding="utf-8", errors="replace")
        except PermissionError as error:
            raise InspectionError("diagnostic_access_denied") from error
        except OSError as error:
            raise InspectionError("diagnostic_unreadable") from error
        matches = RSYNC_CODE_RE.findall(text)
        code = matches[-1] if matches else "NOT_AVAILABLE"
    return [
        ("last_backup_failure_phase", phase),
        ("transfer_exit_code", code),
        ("transfer_exit_code_source", "last_error_log_regex" if code != "NOT_AVAILABLE" else "NOT_AVAILABLE"),
        ("transfer_failure_category", RSYNC_CATEGORIES.get(code, "not_available")),
        ("transfer_started_epoch", "NOT_AVAILABLE"),
        ("transfer_failed_epoch", "NOT_AVAILABLE"),
        ("transfer_duration_seconds", "NOT_AVAILABLE"),
        ("transfer_duration_reason", "wrapper_does_not_persist_transfer_clock"),
    ]


def production(args: argparse.Namespace) -> None:
    backup_root = safe_root(args.backup_root)
    state_dir = safe_root(args.state_dir)
    now = args.now_epoch
    metadata_valid: list[tuple[Path, str, str, BatchResult]] = []
    invalid_count = invalid_bytes = temp_count = temp_bytes = other_count = 0
    for entry in backup_root.iterdir():
        match = BATCH_RE.fullmatch(entry.name)
        if match and entry.is_dir() and not entry.is_symlink():
            result = validate_batch(
                entry,
                match.group(1),
                match.group(2),
                verify_hashes=False,
            )
            if result.valid and result.created_epoch is not None and result.created_epoch <= now:
                metadata_valid.append((entry, match.group(1), match.group(2), result))
            else:
                invalid_count += 1
                invalid_bytes += result.total_bytes
        elif TEMP_RE.fullmatch(entry.name) and entry.is_dir() and not entry.is_symlink():
            temp_count += 1
            temp_bytes += tree_bytes(entry)
        else:
            other_count += 1

    metadata_valid.sort(key=lambda item: item[3].created_epoch or -1)
    latest_metadata = metadata_valid[-1] if metadata_valid else None
    latest_verified: BatchResult | None = None
    if latest_metadata is not None:
        latest_verified = validate_batch(
            latest_metadata[0],
            latest_metadata[1],
            latest_metadata[2],
            verify_hashes=True,
        )
    complete_local_recovery_point = bool(latest_verified and latest_verified.valid)
    latest_result = latest_metadata[3] if latest_metadata else None
    output: list[tuple[str, object]] = [
        ("production_probe_result", "PASS"),
        ("complete_local_recovery_point", complete_local_recovery_point),
        (
            "backup_risk_classification",
            "production_single_point"
            if complete_local_recovery_point
            else "no_proven_recovery_point",
        ),
        ("fully_verified_local_batch_count", 1 if complete_local_recovery_point else 0),
        ("metadata_complete_local_batch_count", len(metadata_valid)),
        (
            "metadata_complete_local_batch_total_bytes",
            sum(item[3].total_bytes for item in metadata_valid),
        ),
        (
            "older_metadata_batch_hash_status",
            "NOT_RUN" if len(metadata_valid) > 1 else "NOT_APPLICABLE",
        ),
        (
            "latest_local_candidate_hash_status",
            "PASS"
            if complete_local_recovery_point
            else ("FAIL" if latest_metadata is not None else "NOT_RUN"),
        ),
        (
            "latest_local_candidate_age_seconds",
            now - latest_result.created_epoch
            if latest_result and latest_result.created_epoch is not None
            else "NOT_AVAILABLE",
        ),
        (
            "latest_local_candidate_bytes",
            latest_result.total_bytes if latest_result else "NOT_AVAILABLE",
        ),
        ("invalid_local_batch_count", invalid_count),
        ("invalid_local_batch_total_bytes", invalid_bytes),
        ("generation_temp_count", temp_count),
        ("generation_temp_total_bytes", temp_bytes),
        ("unclassified_local_entry_count", other_count),
        (
            "transition_recovery_point_action",
            "wait_natural" if complete_local_recovery_point else "request_manual_run",
        ),
    ]
    output.extend(filesystem_fields("production_backup", backup_root))
    output.extend(source_contract(args.wrapper_script, args.backup_script))
    output.extend(parse_transfer_diagnostic(state_dir))
    emit(output)


def stopping_class(samples: list[BatchResult]) -> str:
    if len(samples) < 2:
        return "insufficient_samples"
    sizes = sorted(item.total_bytes for item in samples[-2:])
    difference = sizes[1] - sizes[0]
    relative_limit = sizes[1] * SIMILAR_RELATIVE_PPM // 1_000_000
    similar_limit = min(SIMILAR_ABSOLUTE_BYTES, relative_limit)
    return "similar" if difference <= similar_limit else "different"


def arm(args: argparse.Namespace) -> None:
    inbox_root = safe_root(args.inbox_root)
    archive_root = safe_root(args.archive_root)
    now = args.now_epoch
    partials: list[BatchResult] = []
    other_count = 0
    for entry in inbox_root.iterdir():
        match = PARTIAL_RE.fullmatch(entry.name)
        if match and entry.is_dir() and not entry.is_symlink():
            partials.append(inspect_partial(entry, match.group(1), match.group(2)))
        else:
            other_count += 1
    partials.sort(key=lambda item: item.created_epoch or -1)

    archives: list[BatchResult] = []
    invalid_archive_count = 0
    for entry in archive_root.iterdir():
        match = BATCH_RE.fullmatch(entry.name)
        if match and entry.is_dir() and not entry.is_symlink():
            result = validate_batch(entry, match.group(1), match.group(2))
            if result.valid and result.created_epoch is not None and result.created_epoch <= now:
                archives.append(result)
            else:
                invalid_archive_count += 1
    archives.sort(key=lambda item: item.created_epoch or -1)

    output: list[tuple[str, object]] = [
        ("arm_probe_result", "PASS"),
        ("arm_partial_count", len(partials)),
        ("arm_partial_total_bytes", sum(item.total_bytes for item in partials)),
        ("arm_partial_complete_contract_count", sum(1 for item in partials if item.valid)),
        ("arm_partial_incomplete_contract_count", sum(1 for item in partials if not item.valid)),
        ("arm_partial_stop_offset_class", stopping_class(partials)),
        ("arm_partial_similarity_relative_ppm", SIMILAR_RELATIVE_PPM),
        ("arm_partial_similarity_absolute_bytes", SIMILAR_ABSOLUTE_BYTES),
        ("arm_inbox_other_entry_count", other_count),
        ("arm_valid_atomic_archive_count", len(archives)),
        ("arm_invalid_atomic_archive_count", invalid_archive_count),
        (
            "arm_latest_atomic_archive_age_seconds",
            now - archives[-1].created_epoch if archives else "NOT_AVAILABLE",
        ),
    ]
    for index, sample in enumerate(partials, 1):
        ratio = (
            min(sample.received_artifact_bytes, sample.expected_artifact_bytes)
            * 1_000_000
            // sample.expected_artifact_bytes
            if sample.expected_artifact_bytes and sample.expected_artifact_bytes > 0
            else "NOT_AVAILABLE"
        )
        output.extend(
            [
                (f"arm_partial_sample_{index}_total_bytes", sample.total_bytes),
                (
                    f"arm_partial_sample_{index}_expected_artifact_bytes",
                    sample.expected_artifact_bytes
                    if sample.expected_artifact_bytes is not None
                    else "NOT_AVAILABLE",
                ),
                (f"arm_partial_sample_{index}_received_ratio_ppm", ratio),
                (
                    f"arm_partial_sample_{index}_age_seconds",
                    now - sample.created_epoch
                    if sample.created_epoch is not None
                    else "NOT_AVAILABLE",
                ),
                (f"arm_partial_sample_{index}_complete_contract", sample.valid),
            ]
        )
    output.extend(filesystem_fields("arm_inbox", inbox_root))
    output.extend(filesystem_fields("arm_archive", archive_root))
    emit(output)


def absolute_path(value: str) -> Path:
    path = Path(value)
    if not path.is_absolute():
        raise argparse.ArgumentTypeError("an explicit absolute path is required")
    return path


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="Emit redacted backup-chain incident fields")
    parser.add_argument("--now-epoch", type=int, default=int(datetime.now(timezone.utc).timestamp()))
    subparsers = parser.add_subparsers(dest="mode", required=True)
    production_parser = subparsers.add_parser("production")
    production_parser.add_argument("--backup-root", type=absolute_path, required=True)
    production_parser.add_argument("--state-dir", type=absolute_path, required=True)
    production_parser.add_argument("--wrapper-script", type=absolute_path, required=True)
    production_parser.add_argument("--backup-script", type=absolute_path, required=True)
    production_parser.set_defaults(handler=production)
    arm_parser = subparsers.add_parser("arm")
    arm_parser.add_argument("--inbox-root", type=absolute_path, required=True)
    arm_parser.add_argument("--archive-root", type=absolute_path, required=True)
    arm_parser.set_defaults(handler=arm)
    return parser


def main() -> int:
    args = build_parser().parse_args()
    try:
        args.handler(args)
    except InspectionError as error:
        emit([("probe_result", "FAIL"), ("probe_category", error.category)])
        return 1
    except PermissionError:
        emit([("probe_result", "FAIL"), ("probe_category", "access_denied")])
        return 1
    except (OSError, TypeError, ValueError):
        emit([("probe_result", "FAIL"), ("probe_category", "inspection_io_error")])
        return 1
    except Exception:
        emit([("probe_result", "FAIL"), ("probe_category", "inspection_internal_error")])
        return 1
    return 0


if __name__ == "__main__":
    raise SystemExit(main())