"""Keyed, persistent issue-lock storage (#443) with flock hardening (#438).

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. Acquisition is serialized per issue with ``fcntl.flock``.
"""

from __future__ import annotations

import errno
import fcntl
import json
import os
import re
import tempfile
from contextlib import contextmanager
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._+-]+")


class LockContentionError(RuntimeError):
    """Raised when an exclusive per-issue lock cannot be acquired."""


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 flock_path(json_path: str) -> str:
    return f"{json_path}.lock"


def is_process_alive(pid: int | None) -> bool:
    if not pid or pid <= 0:
        return False
    try:
        os.kill(int(pid), 0)
        return True
    except OSError as exc:
        return exc.errno != errno.ESRCH
    except (TypeError, ValueError):
        return False


@contextmanager
def _exclusive_file_lock(lock_path: str):
    os.makedirs(os.path.dirname(lock_path) or ".", exist_ok=True)
    fd = os.open(lock_path, os.O_CREAT | os.O_RDWR, 0o600)
    try:
        try:
            fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
        except BlockingIOError as exc:
            raise LockContentionError(
                f"could not acquire exclusive lock on '{lock_path}'"
            ) from exc
        yield fd
    finally:
        try:
            fcntl.flock(fd, fcntl.LOCK_UN)
        finally:
            os.close(fd)


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()
    record.setdefault("pid", os.getpid())

    pointer = {
        "pid": os.getpid(),
        "lock_file_path": path,
        "issue_number": issue_number,
        "branch_name": record.get("branch_name"),
        "remote": remote,
        "org": org,
        "repo": repo,
    }
    sentinel = flock_path(path)
    try:
        with _exclusive_file_lock(sentinel):
            existing = read_lock_file(path)
            overwrite_block = assess_foreign_lock_overwrite(existing, record)
            if overwrite_block:
                raise RuntimeError(overwrite_block)
            lease_block = assess_same_issue_lease_conflict(
                existing,
                issue_number=issue_number,
                branch_name=str(record.get("branch_name") or ""),
                worktree_path=str(record.get("worktree_path") or ""),
            )
            if lease_block:
                raise RuntimeError(lease_block)
            save_lock_file(path, record)
            save_lock_file(session_pointer_path(root), pointer)
    except LockContentionError as exc:
        competing = read_lock_file(path)
        if competing:
            owner_pid = competing.get("session_pid") or competing.get("pid")
            raise RuntimeError(
                f"Issue #{issue_number} lock contention: {exc}; competing owner "
                f"pid={owner_pid} (fail closed)"
            ) from exc
        raise RuntimeError(f"Issue #{issue_number} lock contention: {exc} (fail closed)") from exc
    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:
    return assess_lock_freshness(lock, now=now)["live"]


def assess_lock_freshness(
    lock_data: dict[str, Any] | None,
    *,
    now: datetime | None = None,
) -> dict[str, Any]:
    """Classify a lock as live, expired, stale, or absent."""
    current = _lease_now(now)
    if not lock_data:
        return {
            "status": "absent",
            "live": False,
            "stale": False,
            "reason": "no lock record",
        }

    expires_at = lease_expires_at(lock_data)
    lease = lock_data.get("work_lease")
    heartbeat_at = _parse_lease_timestamp(lock_data.get("last_heartbeat_at"))
    if heartbeat_at is None and isinstance(lease, dict):
        heartbeat_at = _parse_lease_timestamp(lease.get("last_heartbeat_at"))

    pid = lock_data.get("session_pid")
    if pid is None:
        pid = lock_data.get("pid")
    pid_alive = is_process_alive(pid) if pid is not None else False

    if expires_at and expires_at <= current:
        return {
            "status": "expired",
            "live": False,
            "stale": True,
            "reason": f"lease expired at {expires_at.isoformat()}",
            "pid_alive": pid_alive,
        }

    if pid is not None and not pid_alive:
        return {
            "status": "stale",
            "live": False,
            "stale": True,
            "reason": f"owner pid {pid} is not alive",
            "pid_alive": False,
        }

    return {
        "status": "live",
        "live": True,
        "stale": False,
        "reason": "lock heartbeat and lease are fresh",
        "pid_alive": pid_alive,
        "heartbeat_at": heartbeat_at.isoformat() if heartbeat_at else None,
        "expires_at": expires_at.isoformat() if expires_at else None,
    }


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


def verify_lock_for_mutation(
    lock_data: dict[str, Any] | None,
    *,
    issue_number: int | None = None,
    branch_name: str | None = None,
    worktree_path: str | None = None,
) -> dict[str, Any]:
    """Re-check lock ownership immediately before a mutation (#438)."""
    reasons: list[str] = []
    if not lock_data:
        return {"proven": False, "block": True, "reasons": ["issue lock is missing (fail closed)"]}

    freshness = assess_lock_freshness(lock_data)
    if not freshness["live"]:
        reasons.append(f"issue lock is not live: {freshness['reason']} (fail closed)")

    if issue_number is not None and lock_data.get("issue_number") != issue_number:
        reasons.append(
            f"issue lock targets #{lock_data.get('issue_number')}, expected #{issue_number} (fail closed)"
        )

    if branch_name is not None and lock_data.get("branch_name") != branch_name:
        reasons.append(
            f"issue lock branch '{lock_data.get('branch_name')}' does not match "
            f"'{branch_name}' (fail closed)"
        )

    if worktree_path is not None:
        locked = os.path.realpath(str(lock_data.get("worktree_path") or ""))
        declared = os.path.realpath(worktree_path)
        if locked != declared:
            reasons.append(
                f"issue lock worktree '{locked}' does not match declared '{declared}' (fail closed)"
            )

    return {
        "proven": not reasons,
        "block": bool(reasons),
        "reasons": reasons,
        "freshness": freshness,
        "lock_proof": format_lock_proof(lock_data, freshness=freshness),
    }


def list_live_locks(
    *,
    lock_dir: str | None = None,
    now: datetime | None = None,
) -> list[dict[str, Any]]:
    """Return live per-issue locks for queue visibility."""
    live: list[dict[str, Any]] = []
    for path in iter_lock_files(lock_dir):
        record = read_lock_file(path)
        if not record:
            continue
        freshness = assess_lock_freshness(record, now=now)
        if not freshness["live"]:
            continue
        live.append(
            {
                "issue_number": record.get("issue_number"),
                "branch_name": record.get("branch_name"),
                "remote": record.get("remote"),
                "org": record.get("org"),
                "repo": record.get("repo"),
                "worktree_path": record.get("worktree_path"),
                "pid": record.get("session_pid") or record.get("pid"),
                "claimant": (
                    record.get("claimant")
                    or (record.get("work_lease") or {}).get("claimant")
                ),
                "freshness": freshness,
                "lock_path": record.get("lock_file_path") or path,
            }
        )
    return live


def format_lock_proof(
    lock_data: dict[str, Any] | None,
    *,
    freshness: dict[str, Any] | None = None,
    competing_live_locks: list[dict[str, Any]] | None = None,
    released: bool | None = None,
) -> str:
    """Canonical issue-lock proof string for final reports."""
    if not lock_data:
        return "issue lock proof: not acquired"
    fresh = freshness or assess_lock_freshness(lock_data)
    owner = lock_data.get("claimant") or {}
    if not owner and isinstance(lock_data.get("work_lease"), dict):
        owner = lock_data["work_lease"].get("claimant") or {}
    parts = [
        "issue lock proof:",
        f"acquired issue #{lock_data.get('issue_number')}",
        f"branch {lock_data.get('branch_name')}",
        f"owner {owner.get('profile') or 'unknown'}",
        f"pid {lock_data.get('session_pid') or lock_data.get('pid')}",
        f"freshness {fresh.get('status')}",
    ]
    if competing_live_locks is not None:
        parts.append(
            "no competing live lock"
            if not competing_live_locks
            else f"competing live locks {len(competing_live_locks)}"
        )
    if released is True:
        parts.append("lock released")
    elif released is False:
        parts.append("lock retained")
    return "; ".join(parts)