#!/usr/bin/env -S bash -c 'exec "$BASH" "$(dirname "$0")/_hermes-python" "$0" "$@"'
"""Render ``tui_gateway/contracts`` into TypeScript and OpenRPC.

Python-only (the Python CI lane has no Node): Pydantic's ``model_json_schema()`` output is walked
by a deliberately small JSON-Schema-subset renderer — object/properties/required, primitives,
enum, const, anyOf-with-null, array/items, ``$ref``, oneOf + discriminator, additionalProperties.
Anything else raises at generation time so an unsupported model is fixed at the model, never
worked around in the output. Prettier runs on the TS when a node_modules binary is present
(output is already in the repo's prettier style; the Python CI lane regenerates and diffs it).
"""

from __future__ import annotations

import json
import re
import subprocess
import sys
from collections import OrderedDict
from pathlib import Path
from typing import Any, get_args, get_type_hints

from pydantic import TypeAdapter
from pydantic.json_schema import GenerateJsonSchema

ROOT = Path(__file__).resolve().parent.parent
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

from tui_gateway import contracts  # noqa: E402,F401  (imports every topic module → fills the tables)
from tui_gateway.contracts.connectors import (  # noqa: E402
    ConnectorAccountStatus,
    ConnectorErrorReason,
    ConnectorToolFacet,
    ConnectorToolsSource,
)
from tui_gateway.contracts.connectors_operation import ConnectionSettleReason, ConnectionTargetState  # noqa: E402
from tui_gateway.contracts.registry import EVENTS, METHODS, SERVER_REQUESTS  # noqa: E402
from tools.connectors.contract import SettleReason, TargetState  # noqa: E402
from tools.connectors.gateway.wire import ConnectionStatus  # noqa: E402
from tools.connectors.portal.tools_cache import ToolsRead  # noqa: E402
from tools.connectors.portal.wire import ConnectorTool  # noqa: E402

TS_OUT = ROOT / "apps" / "shared" / "src" / "gateway-contract.generated.ts"
OPENRPC_OUT = ROOT / "apps" / "shared" / "src" / "gateway-contract.openrpc.json"
_ENUM_PAIRS = (
    ("ConnectionTargetState", ConnectionTargetState, "TargetState", TargetState),
    ("ConnectionSettleReason", ConnectionSettleReason, "SettleReason", SettleReason),
    ("ConnectorAccountStatus", ConnectorAccountStatus, "ConnectionStatus", ConnectionStatus),
    ("ConnectorToolFacet", ConnectorToolFacet, "ConnectorTool.facet", ConnectorTool.model_fields["facet"].annotation),
    ("ConnectorToolsSource", ConnectorToolsSource, "ToolsRead.source", get_type_hints(ToolsRead)["source"]),
)

HEADER = (
    "// GENERATED by scripts/gen_gateway_contracts.py from tui_gateway/contracts — DO NOT EDIT.\n"
    "// Regenerate: .venv/bin/python scripts/gen_gateway_contracts.py\n"
    "// tests/tui_gateway/contracts/test_generated.py fails when this file is stale.\n"
)


class _Schema(GenerateJsonSchema):
    """Stable ``$defs`` naming: the model's class name (no module qualifiers)."""

    def normalize_name(self, name: str) -> str:
        return re.sub(r"[^A-Za-z0-9_]", "_", name)


def _schema_for(models: list[type]) -> tuple[dict[str, dict], list[dict]]:
    """One shared ``$defs`` for every model, plus each model's own schema (a ``$ref`` in practice)."""
    from pydantic.json_schema import models_json_schema

    defs, top = models_json_schema(
        [(m, "serialization") for m in models], schema_generator=_Schema, ref_template="#/$defs/{model}"
    )
    return top.get("$defs", {}), [defs[(m, "serialization")] for m in models]


# ── TypeScript rendering ─────────────────────────────────────────────────────────────────────────


class Renderer:
    def __init__(self, defs: dict[str, dict]):
        self.defs = defs
        self.emitted: OrderedDict[str, str] = OrderedDict()

    def ref_name(self, ref: str) -> str:
        assert ref.startswith("#/$defs/"), ref
        return ref[len("#/$defs/"):]

    def type_of(self, schema: dict, *, inline_depth: int = 0) -> str:
        if "$ref" in schema:
            name = self.ref_name(schema["$ref"])
            self.ensure(name)
            return name
        if "const" in schema:
            return _lit(schema["const"])
        if "enum" in schema:
            return " | ".join(_lit(v) for v in schema["enum"])
        if "anyOf" in schema or "oneOf" in schema:
            variants = schema.get("anyOf") or schema.get("oneOf") or []
            rendered = list(dict.fromkeys(self.type_of(v, inline_depth=inline_depth) for v in variants))
            return " | ".join(rendered)
        t = schema.get("type")
        if isinstance(t, list):
            return " | ".join(self.type_of({**schema, "type": x}, inline_depth=inline_depth) for x in t)
        if t == "string":
            return "string"
        if t in ("integer", "number"):
            return "number"
        if t == "boolean":
            return "boolean"
        if t == "null":
            return "null"
        if t == "array":
            items = schema.get("items")
            if items is None:
                return "unknown[]"
            if "prefixItems" in schema:
                return "[" + ", ".join(self.type_of(x) for x in schema["prefixItems"]) + "]"
            inner = self.type_of(items, inline_depth=inline_depth)
            return f"({inner})[]" if " | " in inner else f"{inner}[]"
        if t == "object" or "properties" in schema or "additionalProperties" in schema:
            return self.object_literal(schema, inline_depth)
        if not schema or set(schema) <= {"title", "description", "default"}:
            return "unknown"
        raise ValueError(f"unsupported JSON-Schema construct: {json.dumps(schema)[:200]}")

    def object_literal(self, schema: dict, depth: int) -> str:
        props = schema.get("properties")
        extra = schema.get("additionalProperties")
        if not props:
            if extra is False:
                return "Record<string, never>"
            if extra in (None, True):
                return "Record<string, unknown>"
            return f"Record<string, {self.type_of(extra, inline_depth=depth + 1)}>"
        required = set(schema.get("required", ()))
        lines = ["{"]
        for key, sub in props.items():
            opt = "" if key in required else "?"
            lines.append(f"  {_prop(key)}{opt}: {self.type_of(sub, inline_depth=depth + 1)}")
        if extra not in (None, False):
            lines.append(f"  [key: string]: {'unknown' if extra is True else self.type_of(extra)}")
        lines.append("}")
        return "\n".join(lines)

    def ensure(self, name: str) -> None:
        if name in self.emitted:
            return
        self.emitted[name] = ""  # cycle guard
        schema = self.defs[name]
        doc = _doc(schema.get("description"))
        if "enum" in schema:
            body = f"export type {name} = {self.type_of({'enum': schema['enum']})}\n"
        elif schema.get("properties"):
            body = f"export interface {name} {self.object_literal(schema, 0)}\n"
        else:
            body = f"export type {name} = {self.type_of(schema)}\n"
        self.emitted[name] = doc + body


_IDENT = re.compile(r"^[A-Za-z_$][A-Za-z0-9_$]*$")


def _prop(key: str) -> str:
    return key if _IDENT.match(key) else _lit(key)


def _const_items(names: list[str]) -> str:
    return ",\n".join(f"  {_lit(n)}" for n in names) + "\n"


def _lit(value) -> str:
    """A TS literal in the repo's prettier style (single quotes) so the committed file needs no
    Node-side formatting pass — the Python CI lane regenerates and diffs it."""
    if isinstance(value, str):
        return "'" + value.replace("\\", "\\\\").replace("'", "\\'") + "'"
    return json.dumps(value)


def _doc(text: str | None, indent: str = "") -> str:
    if not text:
        return ""
    clean = " ".join(text.split())
    return f"{indent}/** {clean} */\n"


def _pascal(name: str) -> str:
    return "".join(p[:1].upper() + p[1:] for p in re.split(r"[._]", name))


def _enum_values(enum) -> set[str]:
    values = getattr(enum, "__members__", None)
    if values is not None:
        return {member.value for member in values.values()}
    if isinstance(enum, str):
        return {enum}
    return {value for member in get_args(enum) for value in _enum_values(member)}


def _check_enum_parity() -> None:
    for contract_name, contract_enum, domain_name, domain_enum in _ENUM_PAIRS:
        contract_values = _enum_values(contract_enum)
        domain_values = _enum_values(domain_enum)
        if contract_values != domain_values:
            raise ValueError(
                f"enum parity failed: {contract_name} vs {domain_name}: "
                f"{sorted(contract_values ^ domain_values)}"
            )


def render_ts() -> str:
    _check_enum_parity()
    models: list[type] = []
    for m in METHODS.values():
        models += [m.params, m.result]
    for r in SERVER_REQUESTS.values():
        models += [r.params, r.result]
    for e in EVENTS.values():
        if e.payload is not None:
            models.append(e.payload)
    # de-dup preserving order
    seen: dict[type, None] = OrderedDict()
    for m in models:
        seen.setdefault(m)
    models = list(seen)
    defs, tops = _schema_for(models)
    r = Renderer(defs)
    name_of = {m: r.ref_name(t["$ref"]) for m, t in zip(models, tops)}
    for m in models:
        r.ensure(name_of[m])

    out = [HEADER, "/* eslint-disable */\n", "// ── Types ──\n"]
    out.extend(r.emitted.values())
    out.append(
        "export type ConnectorErrorReason = "
        + r.type_of({"enum": [member.value for member in ConnectorErrorReason]})
        + "\n"
    )

    out.append("\n// ── Client→server methods ──\n")
    out.append("export interface RpcMethods {\n")
    for m in sorted(METHODS.values(), key=lambda x: x.name):
        out.append(_doc(m.doc, "  "))
        out.append(f"  {_prop(m.name)}: {{ params: {name_of[m.params]}; result: {name_of[m.result]} }}\n")
    out.append("}\n")
    out.append("export type RpcMethod = keyof RpcMethods\n")
    out.append("export const RPC_METHODS = [\n" + _const_items(sorted(METHODS)) + "] as const satisfies readonly RpcMethod[]\n")

    out.append("\n// ── Server→client requests ──\n")
    out.append("export interface ServerRequestMap {\n")
    for s in sorted(SERVER_REQUESTS.values(), key=lambda x: x.name):
        out.append(_doc(s.doc, "  "))
        out.append(f"  {_prop(s.name)}: {{ params: {name_of[s.params]}; result: {name_of[s.result]} }}\n")
    out.append("}\n")
    out.append("export type ServerRequestMethod = keyof ServerRequestMap\n")
    out.append("export const SERVER_REQUEST_METHODS = [\n" + _const_items(sorted(SERVER_REQUESTS))
               + "] as const satisfies readonly ServerRequestMethod[]\n")

    out.append("\n// ── Notifications (`event` frames) ──\n")
    out.append("export interface BackendGatewayEventMap {\n")
    for e in sorted(EVENTS.values(), key=lambda x: x.name):
        out.append(_doc(e.doc, "  "))
        payload = name_of[e.payload] if e.payload is not None else "Record<string, never>"
        out.append(f"  {_prop(e.name)}: {payload}\n")
    out.append("}\n")
    out.append("export type BackendGatewayEventName = keyof BackendGatewayEventMap\n")
    out.append("export const GATEWAY_EVENT_TYPES = [\n" + _const_items(sorted(EVENTS))
               + "] as const satisfies readonly BackendGatewayEventName[]\n")
    return "".join(out)


def _tidy(text: str) -> str:
    """No trailing whitespace, single trailing newline (matches `git diff --check` + prettier)."""
    return "\n".join(line.rstrip() for line in text.splitlines()).rstrip("\n") + "\n"


# ── OpenRPC rendering ────────────────────────────────────────────────────────────────────────────


def _openrpc_schema(model: type) -> dict:
    schema = TypeAdapter(model).json_schema(schema_generator=_Schema, ref_template="#/components/schemas/{model}")
    schema.pop("$defs", None)
    return schema


def render_openrpc() -> str:
    _check_enum_parity()
    components: dict[str, dict] = {}
    all_models: list[type] = []
    for m in METHODS.values():
        all_models += [m.params, m.result]
    for r in SERVER_REQUESTS.values():
        all_models += [r.params, r.result]
    for e in EVENTS.values():
        if e.payload is not None:
            all_models.append(e.payload)
    from pydantic.json_schema import models_json_schema

    seen: dict[type, None] = OrderedDict()
    for m in all_models:
        seen.setdefault(m)
    _, top = models_json_schema(
        [(m, "serialization") for m in seen], schema_generator=_Schema,
        ref_template="#/components/schemas/{model}",
    )
    components = top.get("$defs", {})
    components["ConnectorErrorReason"] = {
        "type": "string",
        "enum": [member.value for member in ConnectorErrorReason],
    }

    def ref(model: type) -> dict:
        return {"$ref": f"#/components/schemas/{model.__name__}"}

    doc = {
        "openrpc": "1.3.2",
        "info": {"title": "Hermes TUI/Desktop gateway", "version": "1",
                 "description": "Generated from tui_gateway/contracts by scripts/gen_gateway_contracts.py."},
        "methods": [
            {"name": m.name, "summary": " ".join(m.doc.split()),
             "params": [{"name": "params", "schema": ref(m.params)}],
             "result": {"name": "result", "schema": ref(m.result)}}
            for m in sorted(METHODS.values(), key=lambda x: x.name)
        ],
        "components": {"schemas": components},
        "x-server-requests": [
            {"name": s.name, "summary": " ".join(s.doc.split()),
             "params": [{"name": "params", "schema": ref(s.params)}],
             "result": {"name": "result", "schema": ref(s.result)}}
            for s in sorted(SERVER_REQUESTS.values(), key=lambda x: x.name)
        ],
        "x-notifications": [
            {"name": e.name, "summary": " ".join(e.doc.split()),
             "params": [{"name": "payload", "schema": ref(e.payload) if e.payload is not None
                         else {"type": "object", "additionalProperties": False}}]}
            for e in sorted(EVENTS.values(), key=lambda x: x.name)
        ],
    }
    return json.dumps(doc, indent=2, sort_keys=False) + "\n"


def render_all() -> dict[Path, str]:
    return {TS_OUT: _tidy(render_ts()), OPENRPC_OUT: render_openrpc()}


def main(argv: list[str] | None = None) -> int:
    args = argv if argv is not None else sys.argv[1:]
    check = "--check" in args
    stale = []
    for path, text in render_all().items():
        current = path.read_text(encoding="utf-8-sig") if path.exists() else None
        if current == text:
            continue
        if check:
            stale.append(path)
        else:
            path.write_text(text, encoding="utf-8")
            print(f"wrote {path.relative_to(ROOT)}")
    if stale:
        for p in stale:
            print(f"stale: {p.relative_to(ROOT)} — run scripts/gen_gateway_contracts.py", file=sys.stderr)
        return 1
    return 0


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