import base64
import contextlib
import json
import os
import re
import subprocess
from datetime import datetime, timedelta, timezone
from typing import Any

from audit_event_reconciliation import redact_sensitive_text
import issue_lock_provenance

DEFAULT_LOCK_TTL_HOURS = 24
AUTHOR_ISSUE_WORK_LEASE = "author_issue_work"

_SANCTIONED_ROLES = {"author", "reviewer", "merger", "reconciler", "controller"}


def default_lock_dir() -> str:
    override = (os.environ.get("GITEA_ISSUE_LOCK_DIR") or "").strip()
    if override:
        return override
    cache_dir = (os.environ.get("GITEA_MCP_SESSION_STATE_DIR") or "").strip()
    if cache_dir:
        return os.path.join(cache_dir, "locks")
    user_cache = os.path.expanduser("~/.cache")
    return os.path.join(user_cache, "gitea-tools", "locks")


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"


@contextlib.contextmanager
def _exclusive_file_lock(lock_path: str):
    import fcntl

    _ensure_lock_dir(os.path.dirname(lock_path))
    with open(lock_path, "a+") as fh:
        try:
            fcntl.flock(fh.fileno(), fcntl.LOCK_EX)
            yield
        finally:
            try:
                fcntl.flock(fh.fileno(), fcntl.LOCK_UN)
            except Exception:
                pass


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()
    slug = f"{remote}-{org}-{repo}-issue-{issue_number}.json"
    safe_slug = re.sub(r"[^A-Za-z0-9._-]", "_", slug)
    return os.path.join(root, safe_slug)


def read_lock_file(path: str) -> dict[str, Any] | None:
    if not path or not os.path.isfile(path):
        return None
    try:
        with open(path, "r", encoding="utf-8") as fh:
            data = json.load(fh)
        if isinstance(data, dict):
            return data
    except Exception:
        pass
    return None


def save_lock_file(path: str, data: dict[str, Any]) -> None:
    _ensure_lock_dir(os.path.dirname(path))
    tmp_path = f"{path}.tmp.{os.getpid()}"
    with open(tmp_path, "w", encoding="utf-8") as fh:
        json.dump(data, fh, indent=2)
        fh.flush()
        os.fsync(fh.fileno())
    os.replace(tmp_path, path)


def lock_generation(lock: dict[str, Any] | None) -> int:
    if not isinstance(lock, dict):
        return 0
    try:
        return int(lock.get("lock_generation") or 0)
    except (TypeError, ValueError):
        return 0


def bind_session_lock(
    lock_data: dict[str, Any],
    lock_dir: str | None = None,
    *,
    expected_generation: int | None = None,
    renewal_sanctioned: bool = False,
    recovery_sanctioned: bool = False,
) -> str:
    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, recovery_sanctioned=recovery_sanctioned
            )
            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 ""),
                renewal_sanctioned=renewal_sanctioned,
                recovery_sanctioned=recovery_sanctioned,
            )
            if lease_block:
                raise RuntimeError(lease_block)

            current_gen = lock_generation(existing)
            if expected_generation is not None and current_gen != expected_generation:
                raise RuntimeError(
                    f"compare-and-swap generation mismatch on issue #{issue_number}: "
                    f"expected {expected_generation}, observed {current_gen} (fail closed)"
                )

            record["lock_generation"] = current_gen + 1
            if not record.get("lock_provenance"):
                provenance_source = (
                    issue_lock_provenance.SOURCE_RECOVER_DIRTY_ORPHANED
                    if recovery_sanctioned
                    else issue_lock_provenance.SOURCE_LOCK_ISSUE
                )
                record["lock_provenance"] = issue_lock_provenance.build_lock_provenance(
                    issue_number=issue_number,
                    source=provenance_source,
                    branch_name=record.get("branch_name"),
                    worktree_path=record.get("worktree_path"),
                    generation=record["lock_generation"],
                )

            save_lock_file(path, record)
            save_lock_file(session_pointer_path(root), pointer)
    except Exception:
        raise
    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 []
    out: list[str] = []
    for entry in os.listdir(root):
        if entry.endswith(".json") and not entry.startswith("session-"):
            out.append(os.path.join(root, entry))
    return sorted(out)


def find_live_lock_for_branch(
    branch_name: str,
    lock_dir: str | None = None,
    *,
    now: datetime | 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
        lock_branch = str(lock.get("branch_name") or "").strip()
        if lock_branch == target and is_lease_live(lock, now=now):
            return lock
    return None


def is_process_alive(pid: int | None) -> bool:
    if pid is None or pid <= 0:
        return False
    try:
        os.kill(pid, 0)
        return True
    except OSError:
        return False


def _lease_now(now: datetime | None = None) -> datetime:
    if now is not None:
        if now.tzinfo is None:
            return now.replace(tzinfo=timezone.utc)
        return now.astimezone(timezone.utc)
    return datetime.now(timezone.utc)


def _parse_lease_timestamp(text: str | None) -> datetime | None:
    if not text:
        return None
    raw = str(text).strip()
    if not raw:
        return None
    try:
        dt = datetime.fromisoformat(raw)
        if dt.tzinfo is None:
            return dt.replace(tzinfo=timezone.utc)
        return dt.astimezone(timezone.utc)
    except Exception:
        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]:
    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")
    if pid is None:
        pid = lock_data.get("owner_pid")
    pid_missing = pid is None or str(pid).strip() == ""
    try:
        pid_int = int(pid) if not pid_missing else None
        if pid_int is not None and pid_int <= 0:
            pid_missing = True
            pid_int = None
    except (TypeError, ValueError):
        pid_missing = True
        pid_int = None
    pid_alive = is_process_alive(pid_int) if pid_int 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,
            "pid_missing": pid_missing,
        }

    if pid_missing:
        return {
            "status": "malformed",
            "live": False,
            "stale": True,
            "reason": (
                "lock has no usable session pid; cannot prove live ownership "
                "(PID-less locks are never live by missing expiry alone)"
            ),
            "pid_alive": False,
            "pid_missing": True,
            "heartbeat_at": heartbeat_at.isoformat() if heartbeat_at else None,
            "expires_at": expires_at.isoformat() if expires_at else None,
        }

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

    raw_status = (lock_data.get("status") or "").strip().lower()
    if raw_status in {"active", "acquired", "locked"}:
        return {
            "status": "active",
            "live": True,
            "stale": False,
            "reason": f"lock active (pid {pid_int})",
            "pid_alive": True,
            "pid_missing": False,
            "pid": pid_int,
        }

    return {
        "status": raw_status or "unknown",
        "live": False,
        "stale": True,
        "reason": f"unrecognized lock status '{raw_status}'",
        "pid_alive": pid_alive,
        "pid_missing": False,
    }


def assess_same_issue_lease_conflict(
    existing_lock: dict[str, Any] | None,
    *,
    issue_number: int,
    branch_name: str,
    worktree_path: str,
    now: datetime | None = None,
    renewal_sanctioned: bool = False,
    recovery_sanctioned: bool = False,
) -> str | None:
    if not existing_lock:
        return None
    freshness = assess_lock_freshness(existing_lock, now=now)
    if not freshness["live"]:
        return None
    if renewal_sanctioned or recovery_sanctioned:
        return None

    existing_wt = str(existing_lock.get("worktree_path") or "").strip()
    target_wt = (worktree_path or "").strip()
    if existing_wt and target_wt:
        try:
            same_wt = os.path.realpath(existing_wt) == os.path.realpath(target_wt)
        except Exception:
            same_wt = existing_wt == target_wt
        if not same_wt:
            owner_pid = existing_lock.get("session_pid") or existing_lock.get("pid")
            return (
                f"Issue #{issue_number} already has an active author_issue_work lease "
                f"from worktree '{existing_wt}' (pid={owner_pid}; fail closed)"
            )

    existing_br = str(existing_lock.get("branch_name") or "").strip()
    target_br = (branch_name or "").strip()
    if existing_br and target_br and existing_br != target_br:
        return (
            f"Issue #{issue_number} already has an active lease on branch '{existing_br}' "
            f"(cannot lock for branch '{target_br}'; fail closed)"
        )
    return None


def assess_foreign_lock_overwrite(
    existing_lock: dict[str, Any] | None,
    proposed_lock: dict[str, Any],
    *,
    now: datetime | None = None,
    recovery_sanctioned: bool = False,
) -> str | None:
    if not existing_lock:
        return None
    freshness = assess_lock_freshness(existing_lock, now=now)

    ex_claimant = (
        existing_lock.get("claimant")
        or existing_lock.get("user")
        or existing_lock.get("username")
        or existing_lock.get("owner")
        or ""
    )
    prop_claimant = (
        proposed_lock.get("claimant")
        or proposed_lock.get("user")
        or proposed_lock.get("username")
        or proposed_lock.get("owner")
        or ""
    )
    ex_profile = (
        existing_lock.get("profile")
        or existing_lock.get("profile_name")
        or existing_lock.get("execution_profile")
        or ""
    )
    prop_profile = (
        proposed_lock.get("profile")
        or proposed_lock.get("profile_name")
        or proposed_lock.get("execution_profile")
        or ""
    )

    same_claimant = bool(
        ex_claimant and prop_claimant and ex_claimant == prop_claimant
    )
    same_profile = bool(
        ex_profile and prop_profile and ex_profile == prop_profile
    )

    if not same_claimant and not recovery_sanctioned:
        return (
            f"Foreign lock overwrite refused: existing lock belongs to claimant "
            f"'{ex_claimant or 'unknown'}' (proposed: '{prop_claimant or 'unknown'}'); "
            f"foreign locks may only be overwritten through explicit sanctioned "
            f"recovery (fail closed)"
        )

    if freshness["live"] and not (same_claimant or same_profile) and not recovery_sanctioned:
        return (
            f"Foreign lock overwrite refused: existing lock is live "
            f"(status={freshness['status']}) and belongs to '{ex_claimant or 'unknown'}'"
        )
    return None