"""Deterministic, bounded updates for repository-owned version metadata."""

from __future__ import annotations

import ast
from dataclasses import dataclass
import json
import os
from pathlib import Path
import re
import tempfile
import tomllib


MAX_VERSION_FILES = 64
MAX_VERSION_FILE_BYTES = 2 * 1024 * 1024


class VersionMetadataError(RuntimeError):
    """Raised when release version metadata cannot be changed safely."""


@dataclass(frozen=True, slots=True)
class VersionFileMutation:
    path: str
    before: bytes
    after: bytes


def version_metadata_mutations(
    repo_path: Path,
    *,
    target_version: str,
) -> tuple[VersionFileMutation, ...]:
    """Render all recognized repository version files without writing them."""

    version = target_version.removeprefix("v")
    candidates: list[tuple[Path, str]] = []
    if (repo_path / "pyproject.toml").is_file():
        candidates.append((repo_path / "pyproject.toml", "pyproject"))
    for package_name, lock_name in (
        ("package.json", "package-lock.json"),
        ("webui/package.json", "webui/package-lock.json"),
        ("webui/package.release.json", "webui/package-lock.release.json"),
    ):
        package = repo_path / package_name
        if not package.is_file():
            continue
        candidates.append((package, "package"))
        lock = repo_path / lock_name
        if lock.is_file():
            candidates.append((lock, "lock"))
    src = repo_path / "src"
    if src.is_dir():
        candidates.extend(
            (path, "manifest")
            for path in sorted(src.glob("**/backend/manifest.py"))
            if path.is_file()
        )
        candidates.extend(
            (path, "package_init")
            for path in sorted(src.glob("*/__init__.py"))
            if path.is_file()
        )
    if len(candidates) > MAX_VERSION_FILES:
        raise VersionMetadataError("Repository version metadata exceeds its file bound.")

    mutations: list[VersionFileMutation] = []
    declarations = 0
    for path, kind in candidates:
        before = _read_bounded(path)
        if kind == "pyproject":
            after, found = _render_pyproject(before, version=version)
        elif kind == "package":
            after, found = _render_package_json(before, version=version)
        elif kind == "lock":
            after, found = _render_package_lock(before, version=version)
        elif kind == "manifest":
            after, found = _render_python_version(
                before,
                path=path,
                target_name="ModuleManifest",
                keyword_name="version",
                version=version,
            )
        else:
            after, found = _render_python_assignment(
                before,
                path=path,
                assignment_name="__version__",
                version=version,
            )
        if not found:
            continue
        declarations += 1
        if before != after:
            mutations.append(
                VersionFileMutation(
                    path=path.relative_to(repo_path).as_posix(),
                    before=before,
                    after=after,
                )
            )
    if declarations == 0:
        raise VersionMetadataError(
            "Repository has no supported version declaration to update."
        )
    return tuple(mutations)


def apply_version_metadata_mutations(
    repo_path: Path,
    *,
    target_version: str,
) -> tuple[str, ...]:
    """Apply one deterministic version update, rolling back on write failure."""

    mutations = version_metadata_mutations(
        repo_path,
        target_version=target_version,
    )
    written: list[VersionFileMutation] = []
    try:
        for mutation in mutations:
            destination = repo_path / mutation.path
            if _read_bounded(destination) != mutation.before:
                raise VersionMetadataError(
                    f"Version metadata changed before update: {mutation.path}"
                )
            _atomic_write(destination, mutation.after)
            written.append(mutation)
    except Exception:
        for mutation in reversed(written):
            _atomic_write(repo_path / mutation.path, mutation.before)
        raise
    return tuple(mutation.path for mutation in mutations)


def version_metadata_paths(
    repo_path: Path,
    *,
    target_version: str,
) -> tuple[str, ...]:
    """Return the exact paths a deterministic target-version update owns."""

    return tuple(
        mutation.path
        for mutation in version_metadata_mutations(
            repo_path,
            target_version=target_version,
        )
    )


def _read_bounded(path: Path) -> bytes:
    try:
        metadata = path.lstat()
    except OSError as exc:
        raise VersionMetadataError(
            f"Version metadata is unavailable: {path.name}"
        ) from exc
    if not path.is_file() or path.is_symlink():
        raise VersionMetadataError(
            f"Version metadata must be a regular file: {path.name}"
        )
    if metadata.st_size > MAX_VERSION_FILE_BYTES:
        raise VersionMetadataError(
            f"Version metadata exceeds its size bound: {path.name}"
        )
    try:
        return path.read_bytes()
    except OSError as exc:
        raise VersionMetadataError(
            f"Version metadata could not be read: {path.name}"
        ) from exc


def _decode(payload: bytes, *, path: Path | None = None) -> str:
    try:
        return payload.decode("utf-8")
    except UnicodeDecodeError as exc:
        label = path.name if path is not None else "file"
        raise VersionMetadataError(
            f"Version metadata is not UTF-8: {label}"
        ) from exc


def _render_pyproject(payload: bytes, *, version: str) -> tuple[bytes, bool]:
    text = _decode(payload)
    try:
        parsed = tomllib.loads(text)
    except tomllib.TOMLDecodeError as exc:
        raise VersionMetadataError("pyproject.toml is malformed.") from exc
    project = parsed.get("project")
    if not isinstance(project, dict) or not isinstance(project.get("version"), str):
        return payload, False
    section = re.search(
        r"(?ms)^\[project\][^\n]*\n(?P<body>.*?)(?=^\[[^\n]+\]|\Z)",
        text,
    )
    if section is None:
        raise VersionMetadataError("pyproject.toml has no editable [project] block.")
    body = section.group("body")
    match = re.search(
        r'(?m)^(?P<prefix>\s*version\s*=\s*)(?P<quote>["\'])(?P<value>[^"\']+)(?P=quote)(?P<suffix>\s*(?:#.*)?)$',
        body,
    )
    if match is None:
        raise VersionMetadataError(
            "pyproject.toml project.version is not a supported literal."
        )
    start = section.start("body") + match.start("value")
    end = section.start("body") + match.end("value")
    rendered = f"{text[:start]}{version}{text[end:]}"
    tomllib.loads(rendered)
    return rendered.encode("utf-8"), True


def _render_package_json(payload: bytes, *, version: str) -> tuple[bytes, bool]:
    data = _json_object(payload, label="package metadata")
    if not isinstance(data.get("version"), str):
        return payload, False
    data["version"] = version
    return _json_bytes(data), True


def _render_package_lock(payload: bytes, *, version: str) -> tuple[bytes, bool]:
    data = _json_object(payload, label="package lock")
    found = False
    if isinstance(data.get("version"), str):
        data["version"] = version
        found = True
    packages = data.get("packages")
    root = packages.get("") if isinstance(packages, dict) else None
    if isinstance(root, dict) and isinstance(root.get("version"), str):
        root["version"] = version
        found = True
    return (_json_bytes(data), True) if found else (payload, False)


def _json_object(payload: bytes, *, label: str) -> dict[str, object]:
    try:
        value = json.loads(_decode(payload))
    except json.JSONDecodeError as exc:
        raise VersionMetadataError(f"{label.capitalize()} is malformed.") from exc
    if not isinstance(value, dict):
        raise VersionMetadataError(f"{label.capitalize()} must be an object.")
    return value


def _json_bytes(value: dict[str, object]) -> bytes:
    return (
        json.dumps(value, indent=2, ensure_ascii=True, separators=(",", ": "))
        + "\n"
    ).encode("utf-8")


def _render_python_version(
    payload: bytes,
    *,
    path: Path,
    target_name: str,
    keyword_name: str,
    version: str,
) -> tuple[bytes, bool]:
    text = _decode(payload, path=path)
    try:
        tree = ast.parse(text, filename=str(path))
    except SyntaxError as exc:
        raise VersionMetadataError(f"Python metadata is malformed: {path.name}") from exc
    values: list[ast.Constant] = []
    for node in ast.walk(tree):
        if not isinstance(node, ast.Call) or _call_name(node.func) != target_name:
            continue
        for keyword in node.keywords:
            if (
                keyword.arg == keyword_name
                and isinstance(keyword.value, ast.Constant)
                and isinstance(keyword.value.value, str)
            ):
                values.append(keyword.value)
    if not values:
        return payload, False
    if len(values) != 1:
        raise VersionMetadataError(
            f"Python metadata has multiple {target_name}.{keyword_name} values: {path.name}"
        )
    return _replace_python_literal(text, values[0], version=version, path=path), True


def _render_python_assignment(
    payload: bytes,
    *,
    path: Path,
    assignment_name: str,
    version: str,
) -> tuple[bytes, bool]:
    text = _decode(payload, path=path)
    try:
        tree = ast.parse(text, filename=str(path))
    except SyntaxError as exc:
        raise VersionMetadataError(f"Python metadata is malformed: {path.name}") from exc
    values: list[ast.Constant] = []
    for node in tree.body:
        if not isinstance(node, (ast.Assign, ast.AnnAssign)):
            continue
        targets = node.targets if isinstance(node, ast.Assign) else [node.target]
        value = node.value
        if (
            any(
                isinstance(target, ast.Name) and target.id == assignment_name
                for target in targets
            )
            and isinstance(value, ast.Constant)
            and isinstance(value.value, str)
        ):
            values.append(value)
    if not values:
        return payload, False
    if len(values) != 1:
        raise VersionMetadataError(
            f"Python metadata has multiple {assignment_name} assignments: {path.name}"
        )
    return _replace_python_literal(text, values[0], version=version, path=path), True


def _replace_python_literal(
    text: str,
    node: ast.Constant,
    *,
    version: str,
    path: Path,
) -> bytes:
    if (
        node.lineno != node.end_lineno
        or node.end_col_offset is None
        or node.col_offset < 0
    ):
        raise VersionMetadataError(
            f"Python version declaration must be a single-line literal: {path.name}"
        )
    lines = text.splitlines(keepends=True)
    line = lines[node.lineno - 1]
    lines[node.lineno - 1] = (
        line[: node.col_offset]
        + json.dumps(version)
        + line[node.end_col_offset :]
    )
    rendered = "".join(lines)
    ast.parse(rendered, filename=str(path))
    return rendered.encode("utf-8")


def _call_name(node: ast.expr) -> str | None:
    if isinstance(node, ast.Name):
        return node.id
    if isinstance(node, ast.Attribute):
        return node.attr
    return None


def _atomic_write(path: Path, payload: bytes) -> None:
    metadata = path.stat()
    descriptor, temporary_name = tempfile.mkstemp(
        prefix=f".{path.name}.",
        dir=path.parent,
    )
    temporary = Path(temporary_name)
    try:
        os.fchmod(descriptor, metadata.st_mode & 0o777)
        with os.fdopen(descriptor, "wb") as handle:
            handle.write(payload)
            handle.flush()
            os.fsync(handle.fileno())
        temporary.replace(path)
        directory = os.open(path.parent, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
        try:
            os.fsync(directory)
        finally:
            os.close(directory)
    except Exception:
        try:
            os.close(descriptor)
        except OSError:
            pass
        temporary.unlink(missing_ok=True)
        raise