# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. =========
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. =========

from __future__ import annotations

import hashlib
import logging
import threading
import weakref
from pathlib import Path, PurePosixPath
from typing import Literal

import httpx

from app.run_context import RunContext, get_current_run_context
from app.service.task import get_task_lock_if_exists

logger = logging.getLogger("space_overlay")

HASH_CHUNK_SIZE = 1024 * 1024

_PATH_LOCKS: weakref.WeakValueDictionary[
    tuple[str, str, str, str], threading.Lock
] = weakref.WeakValueDictionary()
_PATH_LOCKS_GUARD = threading.Lock()
_OVERLAY_SYNC_FAILURES = 0
_OVERLAY_SYNC_FAILURES_GUARD = threading.Lock()


def normalize_server_url(server_url: str | None) -> str:
    if not server_url:
        return ""
    trimmed = server_url.rstrip("/")
    if trimmed.endswith("/api/v1"):
        return trimmed
    return f"{trimmed}/api/v1"


def sha256_of_file(path: Path) -> str | None:
    if not path.exists():
        return None
    if path.is_symlink() or not path.is_file():
        raise ValueError(f"Cannot hash non-regular file: {path}")
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        while chunk := handle.read(HASH_CHUNK_SIZE):
            digest.update(chunk)
    return digest.hexdigest()


def normalize_relative_path(path: str) -> str:
    normalized = PurePosixPath(path.replace("\\", "/"))
    if (
        not normalized.parts
        or normalized.is_absolute()
        or ".." in normalized.parts
    ):
        raise ValueError("Invalid relative path")
    return str(normalized)


def path_write_lock(
    space_id: str,
    project_id: str,
    run_id: str,
    rel_path: str,
) -> threading.Lock:
    """Return the per-run/path writer lock.

    The lock cache is weakly held to avoid unbounded growth. Callers must keep
    the returned lock strongly referenced for the whole critical section,
    preferably as `with path_write_lock(...):`.
    """

    key = (space_id, project_id, run_id, rel_path)
    with _PATH_LOCKS_GUARD:
        lock = _PATH_LOCKS.get(key)
        if lock is None:
            lock = threading.Lock()
            _PATH_LOCKS[key] = lock
        return lock


def overlay_sync_failure_count() -> int:
    with _OVERLAY_SYNC_FAILURES_GUARD:
        return _OVERLAY_SYNC_FAILURES


def _record_overlay_sync_failure(
    *,
    reason: str,
    context: RunContext,
    rel_path: str,
    error_message: str,
) -> None:
    global _OVERLAY_SYNC_FAILURES
    with _OVERLAY_SYNC_FAILURES_GUARD:
        _OVERLAY_SYNC_FAILURES += 1
        failure_count = _OVERLAY_SYNC_FAILURES
    logger.error(
        "space_overlay_sync_failed",
        extra={
            "overlay_reason": reason,
            "overlay_space_id": context.space_id,
            "overlay_project_id": context.project_id,
            "overlay_run_id": context.run_id,
            "overlay_path": rel_path,
            "overlay_failure_count": failure_count,
            "overlay_error_message": error_message,
        },
    )


def run_context_for_task(api_task_id: str) -> RunContext | None:
    context = get_current_run_context()
    if context is not None:
        return context
    task_lock = get_task_lock_if_exists(api_task_id)
    return getattr(task_lock, "run_context", None) if task_lock else None


def relative_to_workdir(
    context: RunContext, path: str | Path
) -> tuple[str, Path] | None:
    workdir = context.working_directory.expanduser().resolve()
    target = Path(path).expanduser()
    if not target.is_absolute():
        target = workdir / target
    target = target.resolve()
    try:
        rel = target.relative_to(workdir)
    except ValueError:
        return None
    return normalize_relative_path(rel.as_posix()), target


def relative_to_artifact_root(
    context: RunContext, path: str | Path
) -> str | None:
    """Return portable identity only for paths owned by the active Run.

    Artifact finalization scans the task output root first and the workspace
    root second. Mirror that boundary here so realtime write events can carry
    the same relative identity without exposing an absolute local path as
    portable identity.
    """

    target = Path(path).expanduser()
    if not target.is_absolute():
        target = context.working_directory.expanduser().resolve() / target
    target = target.resolve()
    roots = (
        context.task_output_root.expanduser().resolve(),
        context.working_directory.expanduser().resolve(),
    )
    seen_roots: set[Path] = set()
    for root in roots:
        if root in seen_roots:
            continue
        seen_roots.add(root)
        try:
            relative_path = target.relative_to(root).as_posix()
            return normalize_relative_path(relative_path)
        except ValueError:
            continue
    return None


def should_record_overlay(context: RunContext, target: Path) -> bool:
    if not context.server_url or not context.auth_header:
        return False
    if context.workdir_mode in {"direct-write", "artifact-only"}:
        return False
    if (
        context.working_directory.resolve()
        == context.task_output_root.resolve()
    ):
        return False
    try:
        target.relative_to(context.task_output_root.expanduser().resolve())
        return False
    except ValueError:
        return True


def post_overlay_write(
    context: RunContext,
    rel_path: str,
    target_path: Path,
    *,
    base_hash: str | None,
    status: Literal["added", "modified", "deleted"],
    file_hash: str | None = None,
    size: int | None = None,
    mode: int | None = None,
) -> bool:
    if not should_record_overlay(context, target_path):
        return True

    server_url = normalize_server_url(context.server_url)
    if not server_url:
        return True

    if status == "deleted":
        file_hash = None
    elif file_hash is None:
        file_hash = sha256_of_file(target_path)
    if (size is None or mode is None) and target_path.exists():
        stat_result = target_path.stat()
        size = stat_result.st_size if size is None else size
        mode = stat_result.st_mode if mode is None else mode
    payload = {
        "run_id": context.run_id,
        "path": rel_path,
        "status": status,
        "hash": file_hash,
        "base_hash": base_hash,
        "base_snapshot_id": context.extra_env.get("baseSnapshotId"),
        "size": size,
        "mode": mode,
        "source_path": str(target_path),
        "source_root": str(context.working_directory.expanduser().resolve()),
        "metadata": {},
    }
    url = (
        f"{server_url}/spaces/{context.space_id}/projects/"
        f"{context.project_id}/overlays"
    )
    headers = {"Authorization": context.auth_header}
    if context.user_id:
        headers["X-User-ID"] = context.user_id

    try:
        with httpx.Client(timeout=5.0) as client:
            response = client.post(url, json=payload, headers=headers)
            if response.is_error:
                _record_overlay_sync_failure(
                    reason=f"http_{response.status_code}",
                    context=context,
                    rel_path=rel_path,
                    error_message=response.text[:500],
                )
                return False
            return True
    except Exception as exc:  # noqa: BLE001 - overlay sync must not fail the tool write.
        _record_overlay_sync_failure(
            reason="exception",
            context=context,
            rel_path=rel_path,
            error_message=str(exc),
        )
        return False