"""Keyed, persistent issue-lock storage (#443).

Replaces the single global ``/tmp/gitea_issue_lock.json`` slot with per-issue
lock files under ``GITEA_ISSUE_LOCK_DIR`` (default
``~/.cache/gitea-tools/issue-locks``). Each MCP session binds its active lock
via a per-process pointer file so concurrent repos/issues never clobber each
other.
"""

from __future__ import annotations

import json
import os
import re
import tempfile
from datetime import datetime, timedelta, timezone
from typing import Any

LOCK_DIR_ENV = "GITEA_ISSUE_LOCK_DIR"
DEFAULT_LOCK_DIR = os.path.expanduser("~/.cache/gitea-tools/issue-locks")
WORK_LEASE_TTL_HOURS = 4
AUTHOR_ISSUE_WORK_LEASE = "author_issue_work"

_SAFE_SEGMENT_RE = re.compile(r"[^A-Za-z0-9._+-]+")


def default_lock_dir() -> str:
    raw = (os.environ.get(LOCK_DIR_ENV) or DEFAULT_LOCK_DIR).strip()
    return raw or DEFAULT_LOCK_DIR


def _sanitize_segment(value: str) -> str:
    text = (value or "").strip()
    if not text:
        return "_"
    return _SAFE_SEGMENT_RE.sub("_", text)


def lock_key(
    *,
    remote: str,
    org: str,
    repo: str,
    issue_number: int,
) -> str:
    return "-".join(
        _sanitize_segment(part)
        for part in (remote, org, repo, str(issue_number))
    )


def lock_file_path(
    *,
    remote: str,
    org: str,
    repo: str,
    issue_number: int,
    lock_dir: str | None = None,
) -> str:
    root = (lock_dir or default_lock_dir()).strip()
    return os.path.join(root, f"{lock_key(remote=remote, org=org, repo=repo, issue_number=issue_number)}.json")


def session_pointer_path(lock_dir: str | None = None) -> str:
    root = (lock_dir or default_lock_dir()).strip()
    return os.path.join(root, f"session-{os.getpid()}.json")


def _ensure_lock_dir(lock_dir: str | None = None) -> str:
    root = (lock_dir or default_lock_dir()).strip()
    os.makedirs(root, mode=0o700, exist_ok=True)
    return root


def read_lock_file(path: str) -> dict[str, Any] | None:
    lock_path = (path or "").strip()
    if not lock_path or not os.path.exists(lock_path):
        return None
    try:
        with open(lock_path, encoding="utf-8") as handle:
            data = json.load(handle)
    except (OSError, json.JSONDecodeError):
        return None
    return data if isinstance(data, dict) else None


def save_lock_file(path: str, data: dict[str, Any]) -> None:
    lock_path = (path or "").strip()
    if not lock_path:
        raise ValueError("lock path is required (fail closed)")
    parent = os.path.dirname(lock_path) or "."
    os.makedirs(parent, mode=0o700, exist_ok=True)
    payload = json.dumps(data, indent=2, sort_keys=True) + "\n"
    fd, temp_path = tempfile.mkstemp(prefix=".lock-", suffix=".json", dir=parent)
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as handle:
            handle.write(payload)
            handle.flush()
            os.fsync(handle.fileno())
        os.replace(temp_path, lock_path)
    finally:
        if os.path.exists(temp_path):
            try:
                os.remove(temp_path)
            except OSError:
                pass


def bind_session_lock(lock_data: dict[str, Any], lock_dir: str | None = None) -> str:
    """Persist a keyed lock and bind it to the current process session."""
    remote = str(lock_data.get("remote") or "")
    org = str(lock_data.get("org") or "")
    repo = str(lock_data.get("repo") or "")
    issue_number = int(lock_data.get("issue_number") or 0)
    if not remote or not org or not repo or issue_number <= 0:
        raise ValueError("lock record must include remote, org, repo, and issue_number")

    root = _ensure_lock_dir(lock_dir)
    path = lock_file_path(
        remote=remote,
        org=org,
        repo=repo,
        issue_number=issue_number,
        lock_dir=root,
    )
    record = dict(lock_data)
    record["lock_file_path"] = path
    record["session_pid"] = os.getpid()
    save_lock_file(path, record)

    pointer = {
        "pid": os.getpid(),
        "lock_file_path": path,
        "issue_number": issue_number,
        "branch_name": record.get("branch_name"),
        "remote": remote,
        "org": org,
        "repo": repo,
    }
    save_lock_file(session_pointer_path(root), pointer)
    return path


def read_session_issue_lock(lock_dir: str | None = None) -> dict[str, Any] | None:
    root = (lock_dir or default_lock_dir()).strip()
    pointer = read_lock_file(session_pointer_path(root))
    if not pointer:
        return None
    lock_path = str(pointer.get("lock_file_path") or "").strip()
    if not lock_path:
        return None
    return read_lock_file(lock_path)


def load_issue_lock(
    *,
    remote: str,
    org: str,
    repo: str,
    issue_number: int,
    lock_dir: str | None = None,
) -> dict[str, Any] | None:
    return read_lock_file(
        lock_file_path(
            remote=remote,
            org=org,
            repo=repo,
            issue_number=issue_number,
            lock_dir=lock_dir,
        )
    )


def iter_lock_files(lock_dir: str | None = None) -> list[str]:
    root = (lock_dir or default_lock_dir()).strip()
    if not os.path.isdir(root):
        return []
    paths: list[str] = []
    for name in os.listdir(root):
        if not name.endswith(".json") or name.startswith("session-"):
            continue
        paths.append(os.path.join(root, name))
    return sorted(paths)


def find_lock_for_branch(
    *,
    remote: str,
    org: str,
    repo: str,
    branch_name: str,
    lock_dir: str | None = None,
) -> dict[str, Any] | None:
    target = (branch_name or "").strip()
    if not target:
        return None
    for path in iter_lock_files(lock_dir):
        lock = read_lock_file(path)
        if not lock:
            continue
        if (
            str(lock.get("remote") or "") == remote
            and str(lock.get("org") or "") == org
            and str(lock.get("repo") or "") == repo
            and str(lock.get("branch_name") or "").strip() == target
        ):
            lock = dict(lock)
            lock.setdefault("lock_file_path", path)
            return lock
    return None


def _lease_now(now: datetime | None = None) -> datetime:
    return now or datetime.now(timezone.utc)


def _parse_lease_timestamp(value: str | None) -> datetime | None:
    text = (value or "").strip()
    if not text:
        return None
    try:
        return datetime.fromisoformat(text.replace("Z", "+00:00")).astimezone(timezone.utc)
    except ValueError:
        return None


def lease_expires_at(lock: dict[str, Any] | None) -> datetime | None:
    if not lock:
        return None
    lease = lock.get("work_lease")
    if not isinstance(lease, dict):
        return None
    return _parse_lease_timestamp(lease.get("expires_at"))


def is_lease_expired(lock: dict[str, Any] | None, *, now: datetime | None = None) -> bool:
    expires = lease_expires_at(lock)
    if expires is None:
        return False
    return expires <= _lease_now(now)


def is_lease_live(lock: dict[str, Any] | None, *, now: datetime | None = None) -> bool:
    if not lock:
        return False
    lease = lock.get("work_lease")
    if not isinstance(lease, dict):
        return True
    expires = _parse_lease_timestamp(lease.get("expires_at"))
    if expires is None:
        return True
    return expires > _lease_now(now)


def _same_realpath(left: str | None, right: str | None) -> bool:
    if not left or not right:
        return False
    try:
        return os.path.realpath(left) == os.path.realpath(right)
    except OSError:
        return left == right


def assess_same_issue_lease_conflict(
    existing_lock: dict[str, Any] | None,
    *,
    issue_number: int,
    branch_name: str,
    worktree_path: str,
    operation_type: str = AUTHOR_ISSUE_WORK_LEASE,
    now: datetime | None = None,
) -> str | None:
    """Return a fail-closed error when a competing live lease blocks acquisition."""
    if not existing_lock:
        return None

    existing_issue = existing_lock.get("issue_number")
    lease = existing_lock.get("work_lease")
    existing_operation = (
        lease.get("operation_type")
        if isinstance(lease, dict)
        else AUTHOR_ISSUE_WORK_LEASE
    )
    if existing_issue != issue_number or existing_operation != operation_type:
        return None

    existing_branch = existing_lock.get("branch_name")
    existing_worktree = existing_lock.get("worktree_path")
    same_owner = (
        existing_branch == branch_name
        and _same_realpath(str(existing_worktree or ""), worktree_path)
    )
    if is_lease_expired(existing_lock, now=now):
        return (
            f"Issue #{issue_number} has an expired {operation_type} lease on "
            f"branch '{existing_branch}' from worktree '{existing_worktree}'. "
            "Recovery review is required before takeover (fail closed)"
        )
    if same_owner:
        return None
    return (
        f"Issue #{issue_number} already has an active {operation_type} lease on "
        f"branch '{existing_branch}' from worktree '{existing_worktree}' "
        "(fail closed)"
    )


def assess_foreign_lock_overwrite(
    existing_lock: dict[str, Any] | None,
    incoming_lock: dict[str, Any],
    *,
    now: datetime | None = None,
) -> str | None:
    """Block writes that would clobber an unrelated live lease on the same key."""
    if not existing_lock:
        return None

    same_issue = existing_lock.get("issue_number") == incoming_lock.get("issue_number")
    same_branch = existing_lock.get("branch_name") == incoming_lock.get("branch_name")
    same_worktree = _same_realpath(
        str(existing_lock.get("worktree_path") or ""),
        str(incoming_lock.get("worktree_path") or ""),
    )
    if same_issue and same_branch and same_worktree:
        return None
    if not is_lease_live(existing_lock, now=now):
        return None
    return (
        "Refusing to overwrite a live foreign issue lock "
        f"(issue #{existing_lock.get('issue_number')}, "
        f"branch '{existing_lock.get('branch_name')}', "
        f"worktree '{existing_lock.get('worktree_path')}') (fail closed)"
    )


def find_live_lock_for_branch(
    branch_name: str,
    lock_dir: str | None = None,
) -> dict[str, Any] | None:
    target = (branch_name or "").strip()
    if not target:
        return None
    for path in iter_lock_files(lock_dir):
        lock = read_lock_file(path)
        if not lock:
            continue
        if str(lock.get("branch_name") or "").strip() != target:
            continue
        if not is_lease_live(lock):
            continue
        record = dict(lock)
        record.setdefault("lock_file_path", path)
        return record
    return None


def resolve_locked_branch_for_session(
    branch_name: str | None = None,
    lock_dir: str | None = None,
) -> str:
    if branch_name:
        lock = find_live_lock_for_branch(branch_name, lock_dir)
        if lock:
            return str(lock.get("branch_name") or "")
    lock = read_session_issue_lock(lock_dir)
    return str((lock or {}).get("branch_name") or "")


def has_active_issue_lock(
    branch: str,
    *,
    lock_dir: str | None = None,
) -> bool:
    target = (branch or "").strip()
    if not target:
        return False
    for path in iter_lock_files(lock_dir):
        lock = read_lock_file(path)
        if not lock:
            continue
        if str(lock.get("branch_name") or "").strip() != target:
            continue
        if is_lease_live(lock):
            return True
    return False