#!/usr/bin/env python3
"""Conservative OpenAPI JSON breaking-change detector for CI.

Exit codes: 0=no breaking findings, 2=breaking findings, 1=input error.
Only local JSON-pointer $refs are resolved. Manual review is still required.
"""
from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path
from typing import Any

METHODS = {"get", "put", "post", "delete", "options", "head", "patch", "trace"}

def load(path: str) -> dict[str, Any]:
    with Path(path).open(encoding="utf-8") as handle:
        value = json.load(handle)
    if not isinstance(value, dict) or not isinstance(value.get("paths"), dict):
        raise ValueError(f"{path}: expected an OpenAPI JSON object with a paths object")
    return value

def pointer(document: dict[str, Any], ref: str) -> Any:
    if not ref.startswith("#/"):
        return None
    value: Any = document
    for token in ref[2:].split("/"):
        token = token.replace("~1", "/").replace("~0", "~")
        if not isinstance(value, dict) or token not in value:
            return None
        value = value[token]
    return value

def resolve(schema: Any, document: dict[str, Any]) -> Any:
    seen: set[str] = set()
    while isinstance(schema, dict) and isinstance(schema.get("$ref"), str):
        ref = schema["$ref"]
        if ref in seen:
            break
        seen.add(ref)
        target = pointer(document, ref)
        if target is None:
            break
        schema = target
    return schema

def add(findings: list[dict[str, str]], location: str, change: str, detail: str) -> None:
    findings.append({"impact": "breaking", "location": location, "change": change, "detail": detail})

def parameters(path_item: dict[str, Any], operation: dict[str, Any]) -> dict[tuple[str, str], dict[str, Any]]:
    rows: dict[tuple[str, str], dict[str, Any]] = {}
    for source in (path_item.get("parameters", []), operation.get("parameters", [])):
        if isinstance(source, list):
            for item in source:
                if isinstance(item, dict) and isinstance(item.get("name"), str) and isinstance(item.get("in"), str):
                    rows[(item["in"], item["name"])] = item
    return rows

def response_schema(response: Any, document: dict[str, Any]) -> Any:
    response = resolve(response, document)
    if not isinstance(response, dict):
        return None
    content = response.get("content")
    if not isinstance(content, dict):
        return response.get("schema")
    preferred = content.get("application/json")
    if isinstance(preferred, dict):
        return preferred.get("schema")
    for media in content.values():
        if isinstance(media, dict) and "schema" in media:
            return media["schema"]
    return None

def compare_schema(old: Any, new: Any, old_doc: dict[str, Any], new_doc: dict[str, Any], location: str, findings: list[dict[str, str]], seen: set[tuple[int, int]] | None = None) -> None:
    old, new = resolve(old, old_doc), resolve(new, new_doc)
    if not isinstance(old, dict) or not isinstance(new, dict):
        return
    seen = set() if seen is None else seen
    marker = (id(old), id(new))
    if marker in seen:
        return
    seen.add(marker)
    old_type, new_type = old.get("type"), new.get("type")
    if old_type and new_type and old_type != new_type:
        add(findings, location, "schema_type_changed", f"{old_type} -> {new_type}")
        return
    old_enum, new_enum = old.get("enum"), new.get("enum")
    if isinstance(old_enum, list) and isinstance(new_enum, list):
        removed = [value for value in old_enum if value not in new_enum]
        if removed:
            add(findings, location, "enum_values_removed", json.dumps(removed, ensure_ascii=False))
    if old_type == "array" or "items" in old:
        compare_schema(old.get("items"), new.get("items"), old_doc, new_doc, location + "[]", findings, seen)
    old_props = old.get("properties") if isinstance(old.get("properties"), dict) else {}
    new_props = new.get("properties") if isinstance(new.get("properties"), dict) else {}
    for name, schema in old_props.items():
        child = f"{location}.{name}"
        if name not in new_props:
            add(findings, child, "response_property_removed", "Previously documented property is absent")
        else:
            compare_schema(schema, new_props[name], old_doc, new_doc, child, findings, seen)

def compare(old: dict[str, Any], new: dict[str, Any]) -> list[dict[str, str]]:
    findings: list[dict[str, str]] = []
    old_paths, new_paths = old["paths"], new["paths"]
    for path, old_item in old_paths.items():
        if not isinstance(old_item, dict):
            continue
        if path not in new_paths:
            add(findings, path, "path_removed", "Entire API path was removed")
            continue
        new_item = new_paths[path]
        if not isinstance(new_item, dict):
            add(findings, path, "path_removed", "Path is no longer an operation object")
            continue
        for method, old_op in old_item.items():
            if method.lower() not in METHODS or not isinstance(old_op, dict):
                continue
            location = f"{method.upper()} {path}"
            new_op = new_item.get(method)
            if not isinstance(new_op, dict):
                add(findings, location, "operation_removed", "HTTP operation was removed")
                continue
            old_params = parameters(old_item, old_op)
            new_params = parameters(new_item, new_op)
            for key, param in new_params.items():
                if param.get("required") is True and (key not in old_params or old_params[key].get("required") is not True):
                    add(findings, location + f" parameter {key[0]}:{key[1]}", "required_parameter_added", "Clients must now supply this parameter")
            old_body = old_op.get("requestBody")
            new_body = new_op.get("requestBody")
            if isinstance(new_body, dict) and new_body.get("required") is True and not (isinstance(old_body, dict) and old_body.get("required") is True):
                add(findings, location + " requestBody", "request_body_became_required", "Clients must now send a request body")
            old_responses = old_op.get("responses") if isinstance(old_op.get("responses"), dict) else {}
            new_responses = new_op.get("responses") if isinstance(new_op.get("responses"), dict) else {}
            for status, old_response in old_responses.items():
                if status not in new_responses:
                    add(findings, location + f" response {status}", "response_removed", "Previously documented response was removed")
                    continue
                compare_schema(response_schema(old_response, old), response_schema(new_responses[status], new), old, new, location + f" response {status}", findings)
    return sorted(findings, key=lambda row: (row["location"], row["change"]))

def markdown(findings: list[dict[str, str]], old_name: str, new_name: str) -> str:
    lines = ["# OpenAPI breaking-change report", "", f"Compared `{old_name}` -> `{new_name}`.", "", f"- Breaking findings: **{len(findings)}**", "", "| Location | Change | Detail |", "|---|---|---|"]
    if findings:
        lines.extend(f"| `{row['location']}` | {row['change']} | {row['detail']} |" for row in findings)
    else:
        lines.append("| — | No supported breaking changes detected | Manual review is still required |")
    lines += ["", "_Conservative static analysis. Review authentication, constraints, polymorphism, callbacks, and runtime behavior manually._"]
    return "\n".join(lines) + "\n"

def main() -> int:
    parser = argparse.ArgumentParser(description="Detect selected breaking changes between OpenAPI JSON documents.")
    parser.add_argument("old")
    parser.add_argument("new")
    parser.add_argument("--format", choices=("markdown", "json"), default="markdown")
    parser.add_argument("--output")
    parser.add_argument("--allow-breaking", action="store_true")
    args = parser.parse_args()
    try:
        findings = compare(load(args.old), load(args.new))
    except (OSError, json.JSONDecodeError, ValueError) as exc:
        print(f"openapi-breaking-diff: {exc}", file=sys.stderr)
        return 1
    rendered = json.dumps({"breaking_count": len(findings), "findings": findings}, indent=2) + "\n" if args.format == "json" else markdown(findings, args.old, args.new)
    if args.output:
        Path(args.output).write_text(rendered, encoding="utf-8")
    else:
        sys.stdout.write(rendered)
    return 0 if args.allow_breaking or not findings else 2

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