#  Copyright (c) Meta Platforms, Inc. and affiliates.
#  This source code is licensed under both the GPLv2 (found in the COPYING file in the root directory)
#  and the Apache 2.0 License (found in the LICENSE.Apache file in the root directory).

#!/usr/bin/env python3
"""Generate preview C API wrapper fragments from a declarative spec.

This prototype intentionally targets a constrained subset of RocksDB's C API:

- plain field setters/getters
- simple method forwarding wrappers
- Status -> char** errptr wrappers
- simple string metadata accessors

It does not attempt to infer wrappers from arbitrary C++ headers.
"""

from __future__ import annotations

import argparse
import json
from pathlib import Path

ROOT = Path(__file__).resolve().parents[2]
from typing import Iterable


HEADER_BANNER = [
    "// @generated",
    "// -----------------------------------------------------------------------------",
    "// Auto-generated by tools/c_api_gen/generate_c_api.py.",
    "// DO NOT EDIT THIS FILE DIRECTLY.",
    "// Source: tools/c_api_gen/spec.json",
    "// -----------------------------------------------------------------------------",
    "",
]


def format_signature(prefix: str, params: list[str], suffix: str) -> list[str]:
    one_line = f"{prefix}{', '.join(params)}{suffix}"
    if len(one_line) <= 80:
        return [one_line]

    lines = [prefix]
    for i, param in enumerate(params):
        trailing = "," if i + 1 < len(params) else ""
        lines.append(f"    {param}{trailing}")
    lines[-1] = f"{lines[-1]}{suffix}"
    return lines


def c_param_decl(param: dict) -> list[str]:
    kind = param["kind"]
    if kind == "scalar":
        return [f'{param["c_type"]} {param["name"]}']
    if kind == "slice":
        return [f'const char* {param["name"]}', f'size_t {param["len_name"]}']
    raise ValueError(f"Unsupported parameter kind: {kind}")


def flatten_params(params: Iterable[dict], include_errptr: bool = False) -> list[str]:
    result: list[str] = []
    for param in params:
        result.extend(c_param_decl(param))
    if include_errptr:
        result.append("char** errptr")
    return result


def render_decl(fn: dict) -> list[str]:
    kind = fn["kind"]
    if kind == "field_setter":
        params = [
            f'{fn["receiver_type"]} {fn["receiver_name"]}',
            f'{fn["value_type"]} {fn["value_name"]}',
        ]
        return format_signature(
            f'extern ROCKSDB_LIBRARY_API void {fn["name"]}(',
            params,
            ");",
        )
    if kind == "field_getter":
        params = [f'{fn["receiver_type"]} {fn["receiver_name"]}']
        return format_signature(
            f'extern ROCKSDB_LIBRARY_API {fn["return_type"]} {fn["name"]}(',
            params,
            ");",
        )
    if kind == "string_field_getter":
        params = [
            f'{fn["receiver_type"]} {fn["receiver_name"]}',
            f'{fn["len_type"]} {fn["len_name"]}',
        ]
        return format_signature(
            f'extern ROCKSDB_LIBRARY_API const char* {fn["name"]}(',
            params,
            ");",
        )
    if kind == "status_getter":
        params = [
            f'{fn["receiver_type"]} {fn["receiver_name"]}',
            "char** errptr",
        ]
        return format_signature(
            f'extern ROCKSDB_LIBRARY_API void {fn["name"]}(',
            params,
            ");",
        )
    if kind == "void_method":
        return format_signature(
            f'extern ROCKSDB_LIBRARY_API void {fn["name"]}(',
            flatten_params(fn["params"]),
            ");",
        )
    if kind == "status_method":
        return format_signature(
            f'extern ROCKSDB_LIBRARY_API void {fn["name"]}(',
            flatten_params(fn["params"], include_errptr=True),
            ");",
        )
    raise ValueError(f"Unsupported function kind: {kind}")


def render_impl(fn: dict) -> list[str]:
    kind = fn["kind"]
    if kind == "field_setter":
        params = [
            f'{fn["receiver_type"]} {fn["receiver_name"]}',
            f'{fn["value_type"]} {fn["value_name"]}',
        ]
        lines = format_signature(f'void {fn["name"]}(', params, ") {")
        value_expr = fn.get("value_expr", fn["value_name"])
        lines.append(f'  {fn["field"]} = {value_expr};')
        lines.append("}")
        return lines
    if kind == "field_getter":
        params = [f'{fn["receiver_type"]} {fn["receiver_name"]}']
        lines = format_signature(
            f'{fn["return_type"]} {fn["name"]}(', params, ") {"
        )
        lines.append(f'  return {fn["expr"]};')
        lines.append("}")
        return lines
    if kind == "string_field_getter":
        params = [
            f'{fn["receiver_type"]} {fn["receiver_name"]}',
            f'{fn["len_type"]} {fn["len_name"]}',
        ]
        lines = format_signature(f'const char* {fn["name"]}(', params, ") {")
        lines.append(f'  *{fn["len_name"]} = {fn["expr"]}.size();')
        lines.append(f'  return {fn["expr"]}.data();')
        lines.append("}")
        return lines
    if kind == "status_getter":
        params = [
            f'{fn["receiver_type"]} {fn["receiver_name"]}',
            "char** errptr",
        ]
        lines = format_signature(f'void {fn["name"]}(', params, ") {")
        lines.append(f'  SaveError(errptr, {fn["expr"]});')
        lines.append("}")
        return lines
    if kind in ("void_method", "status_method"):
        params = flatten_params(fn["params"], include_errptr=(kind == "status_method"))
        lines = format_signature(f'void {fn["name"]}(', params, ") {")
        call = f'{fn["call_expr"]}({", ".join(fn["method_args"])})'
        if kind == "void_method":
          lines.append(f"  {call};")
        else:
          lines.append(f"  SaveError(errptr, {call});")
        lines.append("}")
        return lines
    raise ValueError(f"Unsupported function kind: {kind}")


def regen_comment_lines(mode: str) -> list[str]:
    if mode == "header":
        return [
            "// To regenerate this file:",
            "//   python3 tools/c_api_gen/generate_c_api.py",
            "//     --spec tools/c_api_gen/spec.json",
            "//     --header-out c_api_gen/c_generated_subset.h.inc",
            "//     --source-out c_api_gen/c_generated_subset.cc.inc",
            "",
        ]
    return [
        "// To regenerate this file:",
        "//   python3 tools/c_api_gen/generate_c_api.py",
        "//     --spec tools/c_api_gen/spec.json",
        "//     --header-out c_api_gen/c_generated_subset.h.inc",
        "//     --source-out c_api_gen/c_generated_subset.cc.inc",
        "",
    ]


def section_comment_lines(
    sections: list[str], header_out: Path, source_out: Path
) -> list[str]:
    lines = [
        "// To regenerate this file:",
        "//   python3 tools/c_api_gen/generate_c_api.py",
    ]
    current = "//     --spec tools/c_api_gen/spec.json"
    for section in sections:
        addition = f" --section {section!r}"
        if len(current) + len(addition) > 78:
            lines.append(current)
            current = f"//     --section {section!r}"
        else:
            current += addition
    lines.append(current)
    try:
        rel_header = header_out.relative_to(ROOT)
    except ValueError:
        rel_header = header_out
    try:
        rel_source = source_out.relative_to(ROOT)
    except ValueError:
        rel_source = source_out
    lines.append("//     --header-out")
    lines.append(f"//     {rel_header}")
    lines.append("//     --source-out")
    lines.append(f"//     {rel_source}")
    lines.append("")
    return lines


def render_sections(spec: dict, mode: str, cmd_comment: list[str] | None = None) -> str:
    lines: list[str] = []
    lines.extend(HEADER_BANNER)
    lines.extend(cmd_comment if cmd_comment is not None else regen_comment_lines(mode))
    for section in spec["sections"]:
        lines.append(f'/* {section["name"]} */')
        lines.append("")
        for fn in section["functions"]:
            render = render_decl if mode == "header" else render_impl
            lines.extend(render(fn))
            lines.append("")
    return "\n".join(lines).rstrip() + "\n"


def write_if_changed(path: Path, content: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    if not path.exists() or path.read_text() != content:
        path.write_text(content)


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--spec", required=True, type=Path)
    parser.add_argument("--header-out", required=True, type=Path)
    parser.add_argument("--source-out", required=True, type=Path)
    parser.add_argument(
        "--section",
        action="append",
        default=[],
        help="Only emit the named section(s) from the spec. May be repeated.",
    )
    parser.add_argument(
        "--check",
        action="store_true",
        help="fail instead of writing if outputs are stale",
    )
    args = parser.parse_args()

    spec = json.loads(args.spec.read_text())
    if args.section:
        allowed = set(args.section)
        spec = {
            **spec,
            "sections": [
                section for section in spec["sections"] if section["name"] in allowed
            ],
        }
    comment = regen_comment_lines("header")
    if args.section:
        comment = section_comment_lines(
            args.section, args.header_out, args.source_out
        )
    header = render_sections(spec, "header", cmd_comment=comment)
    source = render_sections(spec, "source", cmd_comment=comment)

    if args.check:
        expected = {
            args.header_out: header,
            args.source_out: source,
        }
        stale: list[str] = []
        for path, content in expected.items():
            if not path.exists() or path.read_text() != content:
                stale.append(str(path))
        if stale:
            for path in stale:
                print(f"stale: {path}")
            return 1
        return 0

    write_if_changed(args.header_out, header)
    write_if_changed(args.source_out, source)
    return 0


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