"""Cross-process advisory locks for single-owner background work.

The backend and runner each take a single-instance ``fcntl`` lock so only one of each
process runs (``mugi_backend._acquire_single_instance_lock`` /
``mugi_runner._acquire_single_runner_lock``). This module generalises that to a *per
resource* lock so that — even across the distinct backend, MCP server, CLI, and runner
processes that all share ``~/.mugi`` — only one process performs a given expensive,
shared-state rebuild at a time.

Why this is needed: the search-index rebuild, embedding reindex, and corpus digest
schedulers each guard against duplicate work with *process-local* flags (``_rebuilding``,
``_reindex_inflight``, ``_digest_jobs``). Those only dedupe **within** one process; two
processes still race the same ``~/.mugi/search.db`` or corpus manifest. This is the
non-dual-runner half of the lifecycle audit (the single-instance runner lock does not help
here because the racers are *different* process types).

Design:
- ``fcntl`` POSIX record locks release automatically when the holding process dies, so a
  crash cannot leak a permanent lock and wedge all future rebuilds.
- ``try_acquire`` is **non-blocking** and **fail-open**: it returns ``False`` only when
  another process *definitively* holds the lock; any locking-infrastructure error (or a
  platform without ``fcntl``) returns ``True`` so the caller degrades to the prior
  in-process-only behaviour rather than skipping necessary work.
- This is a purely *inter-process* guard. Keep the existing in-process single-flight flags;
  POSIX locks do not conflict within the same process, so intra-process dedupe still relies
  on those flags.
"""
from __future__ import annotations

import errno
import logging
import os
import threading

_log = logging.getLogger(__name__)

try:  # POSIX only; degrade to fail-open on platforms without fcntl.
    import fcntl as _fcntl

    _HAS_FCNTL = True
except Exception:  # pragma: no cover - non-POSIX
    _fcntl = None  # type: ignore[assignment]
    _HAS_FCNTL = False

_LOCK_SUBDIR = "locks"


def _lock_dir() -> str:
    home = os.path.expanduser("~/.mugi")
    path = os.path.join(home, _LOCK_SUBDIR)
    os.makedirs(path, exist_ok=True)
    return path


def _safe_key(resource_key: str) -> str:
    cleaned = "".join(
        c if (c.isalnum() or c in ("-", "_", ".")) else "_" for c in str(resource_key)
    ).strip("._")
    return cleaned or "lock"


class CrossProcessLock:
    """Non-blocking, fail-open single-owner lock for a named shared resource.

    Usage::

        lock = CrossProcessLock("search_index_rebuild")
        if not lock.try_acquire():
            return  # another process owns the rebuild; skip the duplicate
        try:
            ...  # do the work
        finally:
            lock.release()
    """

    def __init__(self, resource_key: str) -> None:
        self._key = _safe_key(resource_key)
        self._fd: int | None = None
        self._held_real_lock = False
        self._guard = threading.Lock()

    @property
    def acquired(self) -> bool:
        """True when this object currently holds a real (kernel-backed) lock."""
        return self._held_real_lock

    def try_acquire(self) -> bool:
        """Return True if the caller may proceed (lock held, or fail-open); False only
        when another process definitively holds the lock."""
        with self._guard:
            if self._fd is not None:
                return True
            if not _HAS_FCNTL:
                return True  # fail-open: no cross-process locking available
            try:
                lock_dir = _lock_dir()
            except OSError as exc:
                _log.debug("process_lock: lock dir unavailable (%s); proceeding unlocked", exc)
                return True
            path = os.path.join(lock_dir, f"{self._key}.lock")
            try:
                fd = os.open(path, os.O_RDWR | os.O_CREAT, 0o600)
            except OSError as exc:
                _log.debug("process_lock: cannot open %s (%s); proceeding unlocked", path, exc)
                return True
            try:
                _fcntl.lockf(fd, _fcntl.LOCK_EX | _fcntl.LOCK_NB)
            except (BlockingIOError, OSError) as exc:
                # EAGAIN/EACCES => definitively held by another process: skip.
                err = getattr(exc, "errno", None)
                if isinstance(exc, BlockingIOError) or err in (errno.EAGAIN, errno.EACCES):
                    try:
                        os.close(fd)
                    except OSError:
                        pass
                    return False
                # Any other lock error => fail-open (proceed unlocked).
                try:
                    os.close(fd)
                except OSError:
                    pass
                _log.debug("process_lock: lockf error on %s (%s); proceeding unlocked", path, exc)
                return True
            try:
                os.lseek(fd, 0, os.SEEK_SET)
                os.ftruncate(fd, 0)
                os.write(fd, str(os.getpid()).encode("ascii"))
            except OSError:
                pass
            self._fd = fd
            self._held_real_lock = True
            return True

    def release(self) -> None:
        with self._guard:
            fd = self._fd
            self._fd = None
            self._held_real_lock = False
            if fd is None:
                return
            if _HAS_FCNTL:
                try:
                    _fcntl.lockf(fd, _fcntl.LOCK_UN)
                except OSError:
                    pass
            try:
                os.close(fd)
            except OSError:
                pass

    def __enter__(self) -> "CrossProcessLock":
        self.try_acquire()
        return self

    def __exit__(self, *_exc: object) -> None:
        self.release()
