"""Per-PR reviewer leases for safe parallel review sessions (#407)."""

from __future__ import annotations

import os
import re
import uuid
from datetime import datetime, timedelta, timezone
from typing import Any

MARKER = "<!-- mcp-review-lease:v1 -->"

_FIELD_RE = re.compile(
    r"^\s*([a-z_]+)\s*:\s*(.+?)\s*$",
    re.IGNORECASE | re.MULTILINE,
)
_FULL_SHA = re.compile(r"^[0-9a-f]{40}$", re.IGNORECASE)

_TERMINAL_PHASES = frozenset({"done", "released", "blocked"})
_ACTIVE_PHASES = frozenset({
    "claimed",
    "validating",
    "approved",
    "request-changes",
    "merging",
    "adopted",
})

DEFAULT_LEASE_TTL_MINUTES = 120
STALE_WARNING_MINUTES = 30
RECLAIMABLE_MINUTES = 60

_SESSION_LEASE: dict[str, Any] | None = None


def _parse_timestamp(value: str | None) -> datetime | None:
    if not value:
        return None
    text = value.strip()
    if text.endswith("Z"):
        text = text[:-1] + "+00:00"
    try:
        parsed = datetime.fromisoformat(text)
    except ValueError:
        return None
    if parsed.tzinfo is None:
        return parsed.replace(tzinfo=timezone.utc)
    return parsed.astimezone(timezone.utc)


def _normalize_sha(value: str | None) -> str | None:
    text = (value or "").strip().lower()
    return text if text and _FULL_SHA.match(text) else None


def _parse_pr_ref(value: str | None) -> int | None:
    digits = re.sub(r"[^\d]", "", value or "")
    return int(digits) if digits.isdigit() else None


def new_session_id() -> str:
    return f"{os.getpid()}-{uuid.uuid4().hex[:12]}"


def format_lease_body(
    *,
    repo: str,
    pr_number: int,
    issue_number: int | None,
    reviewer_identity: str,
    profile: str,
    session_id: str,
    worktree: str,
    phase: str,
    candidate_head: str | None,
    target_branch: str,
    target_branch_sha: str | None,
    last_activity: datetime | None = None,
    expires_at: datetime | None = None,
    blocker: str = "none",
) -> str:
    now = last_activity or datetime.now(timezone.utc)
    expires = expires_at or (now + timedelta(minutes=DEFAULT_LEASE_TTL_MINUTES))
    last_text = now.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace(
        "+00:00", "Z"
    )
    expires_text = expires.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace(
        "+00:00", "Z"
    )
    issue_text = f"#{issue_number}" if issue_number else "none"
    lines = [
        MARKER,
        f"repo: {repo}",
        f"pr: #{pr_number}",
        f"issue: {issue_text}",
        f"reviewer_identity: {reviewer_identity}",
        f"profile: {profile}",
        f"session_id: {session_id}",
        f"worktree: {worktree}",
        f"phase: {phase}",
        f"candidate_head: {candidate_head or 'none'}",
        f"target_branch: {target_branch}",
        f"target_branch_sha: {target_branch_sha or 'none'}",
        f"last_activity: {last_text}",
        f"expires_at: {expires_text}",
        f"blocker: {blocker}",
    ]
    return "\n".join(lines)


def parse_lease_comment(body: str) -> dict[str, Any] | None:
    text = body or ""
    if MARKER not in text:
        return None
    fields: dict[str, str] = {}
    for match in _FIELD_RE.finditer(text):
        fields[match.group(1).strip().lower()] = match.group(2).strip()
    if not fields:
        return None
    return {
        "repo": fields.get("repo"),
        "pr_number": _parse_pr_ref(fields.get("pr")),
        "issue_number": _parse_pr_ref(fields.get("issue")),
        "reviewer_identity": fields.get("reviewer_identity"),
        "profile": fields.get("profile"),
        "session_id": fields.get("session_id"),
        "worktree": fields.get("worktree"),
        "phase": (fields.get("phase") or "").strip().lower() or None,
        "candidate_head": _normalize_sha(fields.get("candidate_head")),
        "target_branch": fields.get("target_branch"),
        "target_branch_sha": _normalize_sha(fields.get("target_branch_sha")),
        "last_activity": fields.get("last_activity"),
        "expires_at": fields.get("expires_at"),
        "blocker": fields.get("blocker"),
        "raw_fields": fields,
    }


def _lease_entries(comments: list[dict], *, pr_number: int) -> list[dict]:
    entries: list[dict] = []
    for comment in comments or []:
        parsed = parse_lease_comment(comment.get("body") or "")
        if not parsed:
            continue
        if parsed.get("pr_number") not in (None, pr_number):
            continue
        entries.append({
            **parsed,
            "comment_id": comment.get("id"),
            "author": (comment.get("user") or {}).get("login") or comment.get("author"),
            "created_at": comment.get("created_at"),
            "updated_at": comment.get("updated_at"),
        })
    return entries


def _lease_expired(lease: dict, *, now: datetime) -> bool:
    expires_at = _parse_timestamp(lease.get("expires_at"))
    return bool(expires_at and expires_at <= now)


def _minutes_since_activity(lease: dict, *, now: datetime) -> float | None:
    last = _parse_timestamp(lease.get("last_activity"))
    if not last:
        return None
    return (now - last).total_seconds() / 60.0


def classify_lease_freshness(lease: dict, *, now: datetime | None = None) -> str:
    """Return active, stale_warning, reclaimable, expired, or terminal."""
    now = now or datetime.now(timezone.utc)
    phase = (lease.get("phase") or "").strip().lower()
    if phase in _TERMINAL_PHASES:
        return "terminal"
    if _lease_expired(lease, now=now):
        return "expired"
    minutes = _minutes_since_activity(lease, now=now)
    if minutes is None:
        return "active"
    if minutes >= RECLAIMABLE_MINUTES:
        return "reclaimable"
    if minutes >= STALE_WARNING_MINUTES:
        return "stale_warning"
    return "active"


def find_active_reviewer_lease(
    comments: list[dict],
    *,
    pr_number: int,
    now: datetime | None = None,
) -> dict[str, Any] | None:
    """Newest non-terminal, unexpired lease for *pr_number*."""
    now = now or datetime.now(timezone.utc)
    for lease in reversed(_lease_entries(comments, pr_number=pr_number)):
        phase = (lease.get("phase") or "").strip().lower()
        if phase in _TERMINAL_PHASES:
            continue
        if _lease_expired(lease, now=now):
            continue
        if phase in _ACTIVE_PHASES or phase:
            lease = dict(lease)
            lease["freshness"] = classify_lease_freshness(lease, now=now)
            return lease
    return None


def assess_acquire_lease(
    comments: list[dict],
    *,
    pr_number: int,
    reviewer_identity: str,
    profile: str,
    session_id: str,
    repo: str,
    issue_number: int | None,
    worktree: str,
    candidate_head: str | None,
    target_branch: str,
    target_branch_sha: str | None,
    pr_merged_or_closed: bool = False,
    now: datetime | None = None,
) -> dict[str, Any]:
    """Fail closed when another session holds an active lease.

    When *pr_merged_or_closed* is true the PR has already merged/closed, so any
    reviewer-lease acquisition or adoption for merge work is moot: fail closed
    with a ``post_merge_moot`` reason and never mint a lease body (#515).
    """
    now = now or datetime.now(timezone.utc)
    reasons: list[str] = []
    existing = find_active_reviewer_lease(comments, pr_number=pr_number, now=now)
    post_merge_moot = bool(pr_merged_or_closed)
    if post_merge_moot:
        reasons.append(
            f"post_merge_moot: PR #{pr_number} is already merged/closed; reviewer "
            "lease adoption for merge is moot (fail closed)"
        )
    if existing:
        owner_session = (existing.get("session_id") or "").strip()
        freshness = existing.get("freshness") or classify_lease_freshness(existing, now=now)
        if owner_session and owner_session != session_id and freshness in {
            "active", "stale_warning"
        }:
            reasons.append(
                f"PR #{pr_number} already has active reviewer lease "
                f"(session_id={owner_session}, phase={existing.get('phase')})"
            )
        elif owner_session and owner_session != session_id and freshness == "reclaimable":
            reasons.append(
                f"PR #{pr_number} lease is reclaimable but still held by "
                f"session_id={owner_session}; explicit reclaim not implemented "
                "(fail closed)"
            )

    if not (reviewer_identity or "").strip():
        reasons.append("reviewer identity required for lease acquisition")
    if not (session_id or "").strip():
        reasons.append("session_id required for lease acquisition")
    if not (worktree or "").strip():
        reasons.append("worktree path required for lease acquisition")

    allowed = not reasons
    body = None
    if allowed:
        body = format_lease_body(
            repo=repo,
            pr_number=pr_number,
            issue_number=issue_number,
            reviewer_identity=reviewer_identity,
            profile=profile,
            session_id=session_id,
            worktree=worktree,
            phase="claimed",
            candidate_head=candidate_head,
            target_branch=target_branch,
            target_branch_sha=target_branch_sha,
            last_activity=now,
        )
    return {
        "acquire_allowed": allowed,
        "reasons": reasons,
        "existing_lease": existing,
        "lease_body": body,
        "session_id": session_id,
        "post_merge_moot": post_merge_moot,
    }


def assess_post_merge_moot_lease(
    comments: list[dict],
    *,
    pr_number: int,
    pr_merged: bool = False,
    pr_state: str | None = None,
    merge_commit_sha: str | None = None,
    now: datetime | None = None,
) -> dict[str, Any]:
    """Assess a reviewer lease left lingering on an already-merged/closed PR (#515).

    Read-first and fail-safe:

    - Only treats a lease as moot when the live PR state is merged/closed.
    - Never proposes touching an *active* lease while the PR is still open
      (``cleanup_allowed`` stays false and a refusal reason is returned).
    - When the PR is merged/closed and a lease is still active, ``cleanup_allowed``
      is true and a terminal ``phase: released`` lease body (``blocker:
      post-merge-moot``) is provided so the moot lease can be neutralised by an
      append-only comment — never by deleting a foreign session's comment, and
      never by adopting or merging.

    Posting the released body makes that lease terminal, so a subsequent call
    finds no active lease and reports nothing left to clean (idempotent).
    """
    now = now or datetime.now(timezone.utc)
    merged_or_closed = bool(pr_merged) or (
        str(pr_state or "").strip().lower() == "closed"
    )
    active = find_active_reviewer_lease(comments, pr_number=pr_number, now=now)
    # The newest lease comment is authoritative: once a terminal marker
    # (released/done/blocked) is the latest entry, the lease is resolved even if
    # an earlier non-terminal comment from the same session still lingers. This
    # keeps post-merge cleanup idempotent.
    entries = _lease_entries(comments, pr_number=pr_number)
    newest = entries[-1] if entries else None
    newest_terminal = bool(newest) and (
        (newest.get("phase") or "").strip().lower() in _TERMINAL_PHASES
    )
    reasons: list[str] = []
    cleanup_allowed = False
    release_body: str | None = None
    is_moot = bool(active) and merged_or_closed and not newest_terminal

    if not merged_or_closed:
        if active:
            reasons.append(
                f"PR #{pr_number} is still open; refusing to touch active reviewer "
                "lease (fail closed)"
            )
        else:
            reasons.append(
                f"PR #{pr_number} is still open; no post-merge lease cleanup applicable"
            )
    elif newest_terminal:
        reasons.append(
            f"PR #{pr_number} reviewer lease already released/terminal; nothing to clean"
        )
    elif active:
        cleanup_allowed = True
        release_body = format_lease_body(
            repo=active.get("repo") or "",
            pr_number=pr_number,
            issue_number=active.get("issue_number"),
            reviewer_identity=active.get("reviewer_identity") or "",
            profile=active.get("profile") or "unknown",
            session_id=active.get("session_id") or "",
            worktree=active.get("worktree") or "",
            phase="released",
            candidate_head=active.get("candidate_head"),
            target_branch=active.get("target_branch") or "master",
            target_branch_sha=active.get("target_branch_sha"),
            last_activity=now,
            blocker="post-merge-moot",
        )
    else:
        reasons.append(
            f"PR #{pr_number} is merged/closed but no active reviewer lease remains; "
            "nothing to clean"
        )

    return {
        "pr_number": pr_number,
        "pr_state": pr_state,
        "pr_merged_or_closed": merged_or_closed,
        "merge_commit_sha": merge_commit_sha,
        "active_lease": active,
        "is_moot": is_moot,
        "cleanup_allowed": cleanup_allowed,
        "release_body": release_body,
        "reasons": reasons,
    }


def record_session_lease(
    lease: dict[str, Any],
    *,
    lease_provenance: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Record the in-session lease mirror for mutation gates.

    *lease_provenance* must be supplied by sanctioned MCP tools (#536). Bare
    manual seeding without provenance cannot satisfy merger/reviewer mutation
    gates.
    """
    global _SESSION_LEASE
    stored = dict(lease)
    if lease_provenance:
        stored["lease_provenance"] = dict(lease_provenance)
    _SESSION_LEASE = stored
    return dict(_SESSION_LEASE)


def clear_session_lease() -> None:
    global _SESSION_LEASE
    _SESSION_LEASE = None


def get_session_lease() -> dict[str, Any] | None:
    return dict(_SESSION_LEASE) if _SESSION_LEASE else None


def assess_mutation_lease_gate(
    *,
    pr_number: int,
    comments: list[dict],
    reviewer_identity: str,
    session_id: str | None,
    mutation: str,
    live_head_sha: str | None,
    pinned_head_sha: str | None,
    now: datetime | None = None,
) -> dict[str, Any]:
    """Reviewer mutations require an owned, current PR lease."""
    now = now or datetime.now(timezone.utc)
    reasons: list[str] = []
    session = get_session_lease()
    active = find_active_reviewer_lease(comments, pr_number=pr_number, now=now)

    if not session:
        reasons.append(
            f"no in-session reviewer lease recorded; acquire via "
            f"gitea_acquire_reviewer_pr_lease or adopt via "
            f"gitea_adopt_merger_pr_lease before {mutation}"
        )
    else:
        import merger_lease_adoption as mla

        if not mla.is_sanctioned_session_lease(session):
            reasons.append(
                "in-session lease lacks sanctioned provenance; manual "
                "_SESSION_LEASE seeding is not canonical proof — use "
                "gitea_acquire_reviewer_pr_lease or gitea_adopt_merger_pr_lease"
            )
    if session and session.get("pr_number") != pr_number:
        reasons.append(
            f"session lease is for PR #{session.get('pr_number')}, not #{pr_number}"
        )
    elif session and (session.get("session_id") or "") != (
        session_id or session.get("session_id")
    ):
        reasons.append("session lease session_id mismatch (fail closed)")

    if active:
        owner = (active.get("session_id") or "").strip()
        if owner and session_id and owner != session_id:
            reasons.append(
                f"active PR lease owned by session_id={owner}; current session "
                f"cannot {mutation}"
            )
        pinned = _normalize_sha(pinned_head_sha)
        live = _normalize_sha(live_head_sha)
        lease_head = active.get("candidate_head")
        if pinned and live and pinned != live:
            reasons.append(
                "PR head changed during lease; stop and re-validate before "
                f"reviewer {mutation}"
            )
        if lease_head and live and lease_head != live:
            reasons.append(
                "live PR head differs from lease candidate_head; refresh lease "
                f"before {mutation}"
            )
        freshness = active.get("freshness") or classify_lease_freshness(active, now=now)
        if freshness in {"expired", "reclaimable"}:
            reasons.append(f"reviewer lease freshness is '{freshness}' (fail closed)")
    else:
        reasons.append(f"no active reviewer lease found on PR #{pr_number}")

    allowed = not reasons
    return {
        "mutation_allowed": allowed,
        "block": not allowed,
        "reasons": reasons,
        "active_lease": active,
        "session_lease": session,
    }


def assess_lease_inventory(
    comments_by_pr: dict[int, list[dict]],
    *,
    now: datetime | None = None,
) -> dict[str, Any]:
    """Summarize lease states across PR comment threads."""
    now = now or datetime.now(timezone.utc)
    active: list[dict] = []
    stale: list[dict] = []
    reclaimable: list[dict] = []
    for pr_number, comments in (comments_by_pr or {}).items():
        lease = find_active_reviewer_lease(comments, pr_number=pr_number, now=now)
        if not lease:
            continue
        freshness = lease.get("freshness") or classify_lease_freshness(lease, now=now)
        entry = {"pr_number": pr_number, "session_id": lease.get("session_id"), "freshness": freshness}
        if freshness == "stale_warning":
            stale.append(entry)
        elif freshness == "reclaimable":
            reclaimable.append(entry)
        else:
            active.append(entry)
    return {
        "active_review_leases": active,
        "stale_review_leases": stale,
        "reclaimable_review_leases": reclaimable,
    }