"""File I/O tool implementations (read, write, delete, patch)."""

from __future__ import annotations

import json
import os
import stat as stat_module
import time
from pathlib import Path
from typing import Any

from butler.core.hashline import (
    format_hash_line,
    format_read_output,
    hashline_read_enabled,
)
from butler.core.read_file_partial import (
    build_large_file_summary,
    format_summary_message,
    read_summary_threshold_lines,
)
from butler.core.read_state import (
    normalize_quotes,
    record_read_state,
    require_read_before_edit,
)
from butler.execution_context import get_current_orchestrator, get_current_session_key
from butler.tools.file_io_ops import (
    maybe_format_after_edit_safe,
    record_edit_path_safe,
    record_read_path_safe,
    tool_json_loud,
    verify_hashline_anchors_safe,
)
from butler.tools.path_safety import check_tool_path, tool_safe_root

MAX_READ_FILE_BYTES = 1024 * 1024
MAX_READ_FILE_LINES = 1000


def _raw_tool_path(path: str | os.PathLike[str], root: Path) -> Path:
    raw_path = Path(str(path or ".")).expanduser()
    if not raw_path.is_absolute():
        raw_path = root / raw_path
    return raw_path


def _read_limited_fd(fd: int) -> tuple[bytes, str]:
    chunks: list[bytes] = []
    total = 0
    while True:
        chunk = os.read(fd, min(65536, MAX_READ_FILE_BYTES + 1 - total))
        if not chunk:
            break
        chunks.append(chunk)
        total += len(chunk)
        if total > MAX_READ_FILE_BYTES:
            return b"", f"File too large: maximum is {MAX_READ_FILE_BYTES} bytes"
    return b"".join(chunks), ""


def _write_all_fd(fd: int, data: bytes) -> None:
    view = memoryview(data)
    written = 0
    while written < len(data):
        n = os.write(fd, view[written:])
        if n == 0:
            raise OSError("short write while writing tool file")
        written += n


def _validate_existing_target_unchanged(
    dir_fd: int,
    name: str,
    expected_stat: os.stat_result,
) -> str:
    flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_NONBLOCK", 0)
    try:
        fd = os.open(name, flags, dir_fd=dir_fd)
    except FileNotFoundError:
        return "Access denied: file changed during validation"
    except OSError as exc:
        return _format_open_error(exc)
    try:
        current_stat = os.fstat(fd)
        if not stat_module.S_ISREG(current_stat.st_mode):
            return "Access denied: only regular files can be written"
        if current_stat.st_nlink > 1:
            return "Access denied: hardlinked files are not allowed"
        if (current_stat.st_dev, current_stat.st_ino) != (
            expected_stat.st_dev,
            expected_stat.st_ino,
        ):
            return "Access denied: file changed during validation"
        return ""
    finally:
        os.close(fd)


def _validate_regular_file_stat(
    stat_result: os.stat_result,
    expected_stat: os.stat_result,
) -> str:
    if (stat_result.st_dev, stat_result.st_ino) != (
        expected_stat.st_dev,
        expected_stat.st_ino,
    ):
        return "Access denied: file changed during validation"
    if not stat_module.S_ISREG(stat_result.st_mode):
        return "Access denied: only regular files can be read"
    if stat_result.st_nlink > 1:
        return "Access denied: hardlinked files are not allowed"
    if stat_result.st_size > MAX_READ_FILE_BYTES:
        return f"File too large: maximum is {MAX_READ_FILE_BYTES} bytes"
    return ""


def _symlink_component_error(
    raw_path: Path,
    root: Path,
    *,
    include_final: bool,
) -> str:
    try:
        relative = raw_path.relative_to(root)
    except ValueError:
        return ""
    current = root
    parts = relative.parts if include_final else relative.parts[:-1]
    for part in parts:
        current = current / part
        try:
            if current.is_symlink():
                return "Access denied: symlinks are not allowed"
        except OSError:
            return "Access denied: path could not be validated"
    return ""


def _format_open_error(exc: OSError) -> str:
    if "Too many levels of symbolic links" in str(exc):
        return "Access denied: symlinks are not allowed"
    return str(exc)


def _read_regular_file_bytes(
    path: str | os.PathLike[str],
    *,
    for_write: bool = False,
) -> tuple[bytes, Path | None, os.stat_result | None, str]:
    root = tool_safe_root()
    safety = check_tool_path(path, for_write=for_write)
    if not safety.allowed:
        return b"", None, None, safety.error
    p = safety.path
    if not p.exists():
        return b"", p, None, f"File not found: {path}"

    raw_path = _raw_tool_path(path, root)
    symlink_error = _symlink_component_error(raw_path, root, include_final=True)
    if symlink_error:
        return b"", p, None, symlink_error

    expected_stat = p.stat()
    flags = (
        os.O_RDONLY
        | getattr(os, "O_NOFOLLOW", 0)
        | getattr(os, "O_NONBLOCK", 0)
    )
    try:
        fd = os.open(raw_path, flags)
    except OSError as exc:
        return b"", p, None, _format_open_error(exc)

    try:
        stat_result = os.fstat(fd)
        validation_error = _validate_regular_file_stat(stat_result, expected_stat)
        if validation_error:
            return b"", p, None, validation_error
        data, read_error = _read_limited_fd(fd)
        if read_error:
            return b"", p, None, read_error
        return data, p, stat_result, ""
    finally:
        os.close(fd)


def _atomic_write_text(
    path: str | os.PathLike[str],
    content: str,
    *,
    expected_stat: os.stat_result | None = None,
) -> tuple[Path | None, str]:
    root = tool_safe_root()
    safety = check_tool_path(path, for_write=True)
    if not safety.allowed:
        return None, safety.error
    p = safety.path
    raw_path = _raw_tool_path(path, root)
    symlink_error = _symlink_component_error(raw_path, root, include_final=raw_path.exists())
    if symlink_error:
        return None, symlink_error

    p.parent.mkdir(parents=True, exist_ok=True)
    raw_parent = raw_path.parent
    symlink_error = _symlink_component_error(raw_parent, root, include_final=True)
    if symlink_error:
        return None, symlink_error
    try:
        parent_stat = raw_parent.stat()
    except OSError as exc:
        return None, str(exc)
    if not stat_module.S_ISDIR(parent_stat.st_mode):
        return None, "Access denied: parent path is not a directory"

    if raw_path.exists():
        try:
            current_stat = raw_path.lstat()
        except OSError as exc:
            return None, str(exc)
        if stat_module.S_ISLNK(current_stat.st_mode):
            return None, "Access denied: symlinks are not allowed"
        if not stat_module.S_ISREG(current_stat.st_mode):
            return None, "Access denied: only regular files can be written"
        if current_stat.st_nlink > 1:
            return None, "Access denied: hardlinked files are not allowed"
        if expected_stat is None:
            expected_stat = current_stat
        elif (
            current_stat.st_dev,
            current_stat.st_ino,
        ) != (expected_stat.st_dev, expected_stat.st_ino):
            return None, "Access denied: file changed during validation"

    parent_flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0)
    try:
        dir_fd = os.open(raw_parent, parent_flags)
    except OSError as exc:
        return None, _format_open_error(exc)

    temp_name = f".{raw_path.name}.butler-tmp-{os.getpid()}-{time.monotonic_ns()}"
    temp_created = False
    try:
        dir_stat = os.fstat(dir_fd)
        if (dir_stat.st_dev, dir_stat.st_ino) != (parent_stat.st_dev, parent_stat.st_ino):
            return None, "Access denied: parent directory changed during validation"
        fd = os.open(
            temp_name,
            os.O_WRONLY | os.O_CREAT | os.O_EXCL,
            0o600,
            dir_fd=dir_fd,
        )
        temp_created = True
        try:
            data = content.encode("utf-8")
            _write_all_fd(fd, data)
            os.fsync(fd)
        finally:
            os.close(fd)
        if expected_stat is not None:
            current_error = _validate_existing_target_unchanged(dir_fd, raw_path.name, expected_stat)
            if current_error:
                return None, current_error
        os.replace(temp_name, raw_path.name, src_dir_fd=dir_fd, dst_dir_fd=dir_fd)
        temp_created = False
        return p, ""
    except OSError as exc:
        return None, str(exc)
    finally:
        if temp_created:
            try:
                os.unlink(temp_name, dir_fd=dir_fd)
            except OSError:
                pass
        os.close(dir_fd)


def _tool_read_file(path: str, offset: int = 1, limit: int = 500, **_: Any) -> str:
    def _run() -> str:
        try:
            off = int(offset)
            lim = int(limit)
        except (TypeError, ValueError):
            return json.dumps({"error": "read_file offset and limit must be integers"})
        if lim < 1 or lim > MAX_READ_FILE_LINES:
            return json.dumps({
                "error": f"read_file limit exceeds maximum ({MAX_READ_FILE_LINES} lines)"
            })
        if off < 1:
            return json.dumps({"error": "read_file offset must be >= 1"})

        data, _p, _stat, error = _read_regular_file_bytes(path)
        if error:
            return json.dumps({"error": error})
        text = data.decode("utf-8", errors="replace")
        lines = text.splitlines()
        if (
            off == 1
            and len(lines) > read_summary_threshold_lines()
            and lim >= 100
        ):
            summary = build_large_file_summary(str(path), lines)
            return str(format_summary_message(summary))
        start = off - 1
        end = start + lim
        selected = lines[start:end]
        if _p is not None:
            body = format_read_output(_p, selected, start + 1)
        else:
            if hashline_read_enabled():
                body = "\n".join(
                    format_hash_line(start + i + 1, line)
                    for i, line in enumerate(selected)
                )
            else:
                body = "\n".join(
                    f"{i + start + 1:6}|{line}" for i, line in enumerate(selected)
                )
        if _p is not None and _stat is not None:
            record_read_state(
                _p,
                _stat,
                data,
                session_key=str(get_current_session_key() or "").strip() or None,
            )
            orch = get_current_orchestrator()
            workspace_root = None
            if orch is not None:
                pm = getattr(orch, "project_manager", None)
                proj = pm.get_current() if pm is not None else None
                if proj is not None:
                    workspace_root = Path(proj.workspace)
            record_read_path_safe(_p, workspace_root=workspace_root)
        return str(body)

    return str(tool_json_loud(_run))


def _tool_write_file(path: str, content: str, **_: Any) -> str:
    def _run() -> str:
        guard = require_read_before_edit(path, for_write=True)
        if guard:
            return json.dumps(guard, ensure_ascii=False)
        p, error = _atomic_write_text(path, content)
        if error:
            return json.dumps({"error": error})
        record_edit_path_safe(p)
        payload: dict[str, Any] = {"success": True, "path": str(p), "bytes": len(content.encode("utf-8"))}
        fmt = maybe_format_after_edit_safe(p)
        if fmt:
            payload["post_edit_format"] = fmt
        return json.dumps(payload)

    return str(tool_json_loud(_run))


def _tool_delete_file(path: str, **_: Any) -> str:
    def _run() -> str:
        _data, p, stat_result, error = _read_regular_file_bytes(path, for_write=True)
        if error:
            return json.dumps({"error": error})
        if p is None or stat_result is None:
            return json.dumps({"error": f"File not found: {path}"})
        if not stat_module.S_ISREG(stat_result.st_mode):
            return json.dumps({"error": "Only regular files can be deleted (not directories)"})

        from butler.project.maturity import delete_allowed_for_path

        allowed, gate_msg = delete_allowed_for_path(p)
        if not allowed:
            return json.dumps({"error": gate_msg, "code": "DELETE_MATURITY_GATE"}, ensure_ascii=False)

        root = tool_safe_root()
        raw_path = _raw_tool_path(path, root)
        symlink_error = _symlink_component_error(raw_path, root, include_final=True)
        if symlink_error:
            return json.dumps({"error": symlink_error})

        try:
            os.unlink(raw_path)
        except OSError as exc:
            return json.dumps({"error": str(exc)})

        return json.dumps({"success": True, "path": str(p), "action": "deleted"})

    return str(tool_json_loud(_run))


def _tool_patch(path: str, old_string: str, new_string: str, **_: Any) -> str:
    def _run() -> str:
        guard = require_read_before_edit(path, for_write=True)
        if guard:
            return json.dumps(guard, ensure_ascii=False)
        data, _p, expected_stat, error = _read_regular_file_bytes(path, for_write=True)
        if error:
            return json.dumps({"error": error})
        if _p is not None:
            mismatch = verify_hashline_anchors_safe(_p, old_string)
            if mismatch:
                return json.dumps(mismatch, ensure_ascii=False)
        text = data.decode("utf-8", errors="replace")
        needle = old_string
        replacement = new_string
        count = text.count(needle)
        fuzzy = False
        if count == 0:
            norm_text = normalize_quotes(text)
            norm_old = normalize_quotes(needle)
            count = norm_text.count(norm_old)
            if count > 0:
                text = norm_text
                needle = norm_old
                replacement = normalize_quotes(replacement)
                fuzzy = True
        if count == 0:
            return json.dumps({
                "error": "old_string not found in file",
                "code": "PATCH_OLD_STRING_NOT_FOUND",
                "hint": (
                    "read_file the path again and copy old_string verbatim from the file "
                    "(indentation, quotes, and trailing newlines must match)."
                ),
            }, ensure_ascii=False)
        if count > 1:
            matches: list[dict[str, Any]] = []
            start = 0
            while len(matches) < 3:
                idx = text.find(needle, start)
                if idx < 0:
                    break
                line_no = text.count("\n", 0, idx) + 1
                line_start = text.rfind("\n", 0, idx) + 1
                line_end = text.find("\n", idx)
                if line_end < 0:
                    line_end = len(text)
                excerpt = text[line_start:line_end].strip()
                if len(excerpt) > 120:
                    excerpt = excerpt[:117] + "..."
                matches.append({"line": line_no, "excerpt": excerpt})
                start = idx + len(needle)
            return json.dumps({
                "error": f"old_string found {count} times; must be unique",
                "match_count": count,
                "matches": matches,
            })
        new_text = text.replace(needle, replacement, 1)
        _written_path, write_error = _atomic_write_text(path, new_text, expected_stat=expected_stat)
        if write_error:
            return json.dumps({"error": write_error})
        record_edit_path_safe(_written_path)
        payload: dict[str, Any] = {"success": True, "replacements": 1}
        if _written_path is not None:
            payload["path"] = str(_written_path)
        if fuzzy:
            payload["fuzzy_quotes"] = True
        if _written_path is not None:
            fmt = maybe_format_after_edit_safe(_written_path)
            if fmt:
                payload["post_edit_format"] = fmt
        return json.dumps(payload)

    return str(tool_json_loud(_run))