"""Library-level process lock, and detection of an incompatible legacy run
(US07-05, concept §15 "migration and operational risks").

Every safety this application has — durable job leases, rename journals, archive
manifests — assumes that one installation owns the library. Two workers, or the
frozen CLI running beside the app, break that assumption *below* the level those
mechanisms can see: the second process simply does not know the first one's
database exists.

So mutation requires a file lock in the data directory, shaped as JSON so any
future or migrated entry point can read and honour it without importing this
package:

    {"lock_version": 1, "role": "worker", "pid": 4242, "host": "...",
     "started_at": "...", "library_roots": ["..."]}

One holder per role: an API and a worker are designed to run together, a second
worker is not. A lock whose process is gone is stale and is taken over — refusing
to start because of a crashed predecessor would turn one outage into two.

Ownership is an advisory ``flock`` on that file, not the record inside it. The
record says *who*; the kernel says *whether*. That distinction is what makes the
lock work in containers (US08-03), where a PID and a hostname are namespaced: a
lock left behind by a container that no longer exists names a pid that still
"exists" in the new container and a host that cannot be probed, so believing the
file would deadlock every restart. A flock is released when its holder dies however
it dies, and is seen by every process that can open the file — which for a local
data directory is every container of this deployment.

Legacy detection is deliberately a heuristic, not a promise: the archived CLI has
no lock of its own, so what can be observed is its state files being written right
now. Recent writes to them mean something else is mutating this library, and every
mutating stage should refuse until it stops.
"""

from __future__ import annotations

import fcntl
import json
import os
import socket
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path

from photo_pipeline.config import Config

LOCK_VERSION = 1
LOCK_SUFFIX = ".lock.json"
# State files only the archived CLIs write. Their presence is history; a *recent*
# modification is a running process.
# ponytail: the real fix is a lock the migrated CLI paths take too — this catches
# the frozen archive, which has no lock and cannot be changed (US07-01).
LEGACY_ARTIFACTS = (
    "photo_analyzer.db",
    "nsfw_scores.csv",
    "photo_analyzer_history.jsonl",
    "photo_analyzer.log",
    "photo_analyzer_debug.log",
)
LEGACY_ACTIVE_SECONDS = 300


class LockHeld(RuntimeError):
    """Another live process of the same role owns this library."""

    def __init__(self, holder: "Holder") -> None:
        super().__init__(
            f"{holder.role} is already running for this library "
            f"(pid {holder.pid} on {holder.host}, since {holder.started_at})"
        )
        self.holder = holder


class LegacyProcessActive(RuntimeError):
    """A legacy CLI appears to be mutating the same library right now."""


@dataclass(frozen=True)
class Holder:
    role: str
    pid: int
    host: str
    started_at: str
    lock_version: int = LOCK_VERSION
    library_roots: tuple[str, ...] = ()

    @property
    def alive(self) -> bool:
        """Whether the recorded process still exists on this host.

        A lock from another host cannot be probed, so it is believed: assuming a
        remote holder is dead is how two machines end up renaming the same folder.
        """
        if self.host != socket.gethostname():
            return True
        try:
            os.kill(self.pid, 0)
        except ProcessLookupError:
            return False
        except PermissionError:
            return True  # exists, owned by someone else
        return True

    def as_dict(self) -> dict:
        return {
            "lock_version": self.lock_version,
            "role": self.role,
            "pid": self.pid,
            "host": self.host,
            "started_at": self.started_at,
            "library_roots": list(self.library_roots),
            "alive": self.alive,
        }


def _now() -> datetime:
    return datetime.now(timezone.utc)


def _in_container() -> bool:
    from photo_pipeline import path_policy

    return path_policy.in_container()


def legacy_activity(config: Config) -> dict:
    """Legacy state files written within the activity window, if any."""
    seen: list[dict] = []
    cutoff = _now().timestamp() - LEGACY_ACTIVE_SECONDS
    roots = [Path(root) for root in config.library_roots] + [Path(config.data_dir)]
    for root in roots:
        for name in LEGACY_ARTIFACTS:
            path = root / name
            try:
                modified = path.stat().st_mtime
            except OSError:
                continue
            if modified >= cutoff:
                seen.append(
                    {
                        "path": str(path),
                        "modified_at": datetime.fromtimestamp(modified, timezone.utc).isoformat(),
                    }
                )
    return {"active": bool(seen), "artifacts": seen, "window_seconds": LEGACY_ACTIVE_SECONDS}


class LibraryLock:
    """One holder per role for one library. Used as a context manager."""

    def __init__(self, config: Config, role: str = "worker") -> None:
        self._config = config
        self.role = role
        self.path = Path(config.data_dir) / f"{role}{LOCK_SUFFIX}"
        self._acquired = False
        self._handle = None

    # ── inspection ────────────────────────────────────────────────────────────

    def holder(self) -> Holder | None:
        try:
            payload = json.loads(self.path.read_text())
        except (OSError, ValueError):
            return None
        try:
            return Holder(
                role=payload["role"],
                pid=int(payload["pid"]),
                host=payload["host"],
                started_at=payload["started_at"],
                lock_version=int(payload.get("lock_version", LOCK_VERSION)),
                library_roots=tuple(payload.get("library_roots", ())),
            )
        except (KeyError, TypeError, ValueError):
            # An unreadable lock is not an absent lock: something wrote it.
            return Holder(role=self.role, pid=-1, host="unknown", started_at="unknown")

    # ── acquire / release ─────────────────────────────────────────────────────

    def acquire(self, *, allow_legacy: bool = False) -> Holder:
        """Take the lock for this role, or explain who has it.

        Raises ``LockHeld`` when a live process of the same role owns the library,
        and ``LegacyProcessActive`` when the archived CLI looks like it is running
        against it.
        """
        if not allow_legacy:
            legacy = legacy_activity(self._config)
            if legacy["active"]:
                raise LegacyProcessActive(
                    "a legacy CLI is writing this library "
                    f"({', '.join(item['path'] for item in legacy['artifacts'])}); "
                    "stop it before running the application"
                )

        self.path.parent.mkdir(parents=True, exist_ok=True)
        # The kernel decides, because the file cannot: a container's PID and hostname
        # are namespaced, so a lock left by a container that no longer exists names a
        # pid that "exists" and a host that cannot be probed (US08-03). An advisory
        # flock is held by a live process or by nobody, is released when that process
        # dies however it dies, and is shared by every process that can open this
        # file — which, for a local data directory, is every role in every container
        # of this deployment.
        handle = open(self.path, "a+", encoding="utf-8")
        try:
            fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
        except OSError:
            handle.close()
            raise LockHeld(self.holder() or Holder(self.role, -1, "unknown", "unknown")) from None

        current = self.holder()
        if current is not None and current.host != socket.gethostname() and not _in_container():
            # We hold the kernel's lock, so nothing on *this* machine holds the file.
            # On a host that still leaves one case open: a data directory shared with
            # another machine, whose flock we cannot trust. Believe its record rather
            # than run two writers. In a container the data volume is local by
            # construction (US08-03), and a foreign hostname is only a dead container.
            fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
            handle.close()
            raise LockHeld(current)

        mine = Holder(
            role=self.role,
            pid=os.getpid(),
            host=socket.gethostname(),
            started_at=_now().isoformat(),
            library_roots=tuple(str(root) for root in self._config.library_roots),
        )
        payload = {k: v for k, v in mine.as_dict().items() if k != "alive"}
        handle.seek(0)
        handle.truncate()
        json.dump(payload, handle, indent=2)
        handle.flush()
        # Held open on purpose: closing it is what releases the lock, and that must
        # happen when this process ends, not when this method returns.
        self._handle = handle
        self._acquired = True
        return mine

    def release(self) -> None:
        """Give up a lock this process owns. Another holder's lock is left alone."""
        if not self._acquired:
            return
        self.path.unlink(missing_ok=True)
        if self._handle is not None:
            self._handle.close()  # closing the descriptor releases the kernel lock
            self._handle = None
        self._acquired = False

    def __enter__(self) -> "LibraryLock":
        self.acquire()
        return self

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