"""Bulk-fetch and lazy-load journal-quality datasets.

This module is now a thin orchestration layer over the
`data_sources` package. Each academic dataset is a `DataSource`
subclass; this module just iterates `ALL_SOURCES` to drive the bulk
download flow (the dashboard "Download Data" button) and to compute the
status payload returned by the `/metrics/api/journal-data/status`
endpoint.

Public API (kept stable for existing callers and tests):
- `JOURNAL_DATA_VERSION`
- `get_journal_data_status()`
- `download_journal_data(force=False)`
- `ensure_journal_data(auto_download=True)`

Test-patch surface (kept stable for tests that mock by string path):
- `_get_data_dir`
- `_fetch_openalex_sources` (re-export shim)
- `_fetch_doaj_journals` (re-export shim)

Data source registry lives in `.data_sources` and is the single source
of truth for source metadata, fetch logic, and download policy.
"""

from __future__ import annotations

import json
import time
from pathlib import Path
from typing import Optional

from loguru import logger

from .data_sources import ALL_SOURCES, get_source

JOURNAL_DATA_VERSION = "v4"

_SENTINEL = ".downloading"

# If the sentinel is older than this, assume the previous download
# crashed mid-way (thread died before the `finally` cleanup could run)
# and reclaim it. The expected wall-clock is ~7 minutes, so 20 minutes
# is a generous safety margin that still catches stuck sentinels.
_SENTINEL_STALE_SECS = 20 * 60

# Shared progress state for the dashboard's status endpoint.
# A module-level dict is deliberate: there's only ever one concurrent
# download (enforced by the O_EXCL sentinel), and the status endpoint
# reads it on a best-effort basis. Structure:
#   state: "idle" | "running" | "error" | "success"
#   started_at: epoch seconds, or None
#   finished_at: epoch seconds, or None
#   sources: {src_key: {name, state, detail, count}}
#   db_build: {state, detail}
#   error_msg: str or None
#
# Per-source entries track independent parallel downloads. Writes are
# atomic at the per-key level in CPython, so workers can update their
# own sub-dict without a lock; the main thread composes the overall
# `state` / `error_msg` after joining.
_download_state: dict = {
    "state": "idle",
    "started_at": None,
    "finished_at": None,
    "sources": {},
    "db_build": {"state": "pending", "detail": ""},
    "error_msg": None,
    # Per-source final counts from the most recently COMPLETED download.
    # Set to a dict only on the success path; explicitly invalidated to
    # None on every other return in `download_journal_data` (up-to-date,
    # disk-space, sentinel-race, required-failure, DB-build-failure) so
    # callers cannot read stale counts from a prior successful run.
    # Callers rendering a user-facing summary should prefer this
    # structured field over parsing the `(success, message)` tuple's
    # string.
    "counts": None,
}


def get_download_state() -> dict:
    """Return a copy of the current download progress state.

    A shallow dict copy is not enough — callers (the status endpoint)
    serialize this to JSON and would otherwise race with live updates.
    We copy the nested dicts too.
    """
    counts = _download_state["counts"]
    return {
        "state": _download_state["state"],
        "started_at": _download_state["started_at"],
        "finished_at": _download_state["finished_at"],
        "db_build": dict(_download_state["db_build"]),
        "error_msg": _download_state["error_msg"],
        "sources": {k: dict(v) for k, v in _download_state["sources"].items()},
        "counts": dict(counts) if counts is not None else None,
    }


def _set_source_state(key: str, **updates) -> None:
    """Update a single source's state entry. Safe to call from worker
    threads because the only writer of ``sources[key]`` is that key's
    own worker (CPython guarantees dict-item writes are atomic).
    """
    entry = _download_state["sources"].setdefault(
        key, {"name": key, "state": "pending", "detail": "", "count": 0}
    )
    entry.update(updates)
    logger.info(
        f"journal-data progress: {entry.get('name', key)} "
        f"{entry.get('state', '?')} {entry.get('detail', '')}".rstrip()
    )


def _get_data_dir() -> Path:
    """Get the journal data directory (user-writable).

    Kept as a module-level function so tests can patch it via
    `mock.patch("...journal_data_downloader._get_data_dir", ...)`.
    """
    from ..config.paths import get_journal_data_directory

    return get_journal_data_directory()


def _clear_orphan_sentinel_on_startup() -> None:
    """Remove any ``.downloading`` sentinel left over from a previous
    process that got killed mid-download.

    The sentinel is created inside ``download_journal_data`` and cleaned
    up in its ``finally`` block. If the process is SIGKILLed (or crashes
    hard enough that the ``finally`` doesn't run) the file sits on disk
    forever. Every subsequent call then sees "Download already in
    progress" and bows out, even though nothing is actually downloading.

    Called once at import time. A fresh process can't possibly own an
    in-progress download, so any pre-existing sentinel is by definition
    an orphan. The 20-minute ``_SENTINEL_STALE_SECS`` recovery path
    still exists for the "process still alive but hung" case.

    Swallows all exceptions — a misread on startup must not break the
    module. Tests that run concurrently in the same process
    (test_concurrent_download_blocked) set the sentinel deliberately
    *after* import, so this runs once and doesn't interfere.
    """
    try:
        sentinel = _get_data_dir() / _SENTINEL
        if sentinel.exists():
            sentinel.unlink()
            logger.warning(
                f"Cleared orphan {_SENTINEL} sentinel from a previous run "
                f"(the old process was killed mid-download; a fresh process "
                f"cannot own an in-progress download)."
            )
    except Exception:
        logger.exception(
            "Could not clear orphan sentinel on startup; "
            "new downloads may be blocked until the 20-minute stale timer."
        )


_clear_orphan_sentinel_on_startup()


# ---------------------------------------------------------------------------
# Test-patch shims for `_fetch_openalex_sources` and `_fetch_doaj_journals`
#
# Tests in `tests/utilities/test_journal_data_downloader.py` patch these
# names by string path. The bodies have moved into `OpenAlexSource.fetch`
# and `DOAJSource.fetch`, but we expose module-level wrappers so the
# existing patches keep intercepting calls. The bulk download loop below
# routes both sources through these wrappers (not directly through
# `.fetch()`) so that `mock.patch` substitutions take effect.
# ---------------------------------------------------------------------------


def _fetch_openalex_sources(data_dir: Path, progress_cb=None) -> int:
    return get_source("openalex").fetch(data_dir, progress_cb=progress_cb)


def _fetch_doaj_journals(data_dir: Path, progress_cb=None) -> int:
    return get_source("doaj").fetch(data_dir, progress_cb=progress_cb)


def _fetch_predatory(data_dir: Path, progress_cb=None) -> int:
    return get_source("predatory").fetch(data_dir, progress_cb=progress_cb)


def _fetch_jabref_abbreviations(data_dir: Path, progress_cb=None) -> int:
    return get_source("jabref").fetch(data_dir, progress_cb=progress_cb)


def _fetch_institutions(data_dir: Path, progress_cb=None) -> int:
    return get_source("institutions").fetch(data_dir, progress_cb=progress_cb)


# Map source key → module-level shim *name*, resolved at call time so
# `mock.patch` substitutions on the module attribute take effect (capturing
# function objects in a dict at import time would defeat patching).
_FETCH_SHIM_NAMES = {
    "openalex": "_fetch_openalex_sources",
    "doaj": "_fetch_doaj_journals",
    "predatory": "_fetch_predatory",
    "jabref": "_fetch_jabref_abbreviations",
    "institutions": "_fetch_institutions",
}


# ---------------------------------------------------------------------------
# Status
# ---------------------------------------------------------------------------


def get_journal_data_status() -> dict:
    """Return the status payload for the dashboard data sources banner.

    Shape (kept stable for existing JS consumers and tests):
        {
          "available": bool,           # OpenAlex source OR compiled DB present
          "version": Optional[str],
          "latest_version": str,
          "needs_update": bool,
          "files": dict[str, bool],    # legacy: filename → present
          "sources": list[dict],       # per-source detail for the banner
          "data_dir": str,
        }
    """
    data_dir = _get_data_dir()
    version_file = data_dir / "version.json"

    installed_version: Optional[str] = None
    if version_file.exists():
        try:
            with open(version_file, encoding="utf-8") as f:
                info = json.load(f)
            installed_version = info.get("version")
        except (json.JSONDecodeError, OSError):
            pass

    files = {src.filename: src.is_present(data_dir) for src in ALL_SOURCES}
    sources = [src.status_dict(data_dir) for src in ALL_SOURCES]

    # Available if the (required) OpenAlex source exists OR the compiled
    # reference DB exists. The compiled DB is useful even when the source
    # JSON has been deleted, since the dashboard can still query it.
    has_source = files.get("openalex_sources.json.gz", False)
    has_db = (data_dir / "journal_quality.db").exists() or (
        data_dir / "journal_reference.db"
    ).exists()

    return {
        "available": has_source or has_db,
        "version": installed_version,
        "latest_version": JOURNAL_DATA_VERSION,
        # `needs_update` is True both when no version is installed at all
        # (first run — show the download CTA) and when the installed
        # version is older than the bundled latest. The previous `and`
        # spelling silently hid the first-run case from the dashboard.
        "needs_update": (
            installed_version is None
            or installed_version != JOURNAL_DATA_VERSION
        ),
        "files": files,
        "sources": sources,
        "data_dir": str(data_dir),
        # Live progress for the dashboard's status indicator. The client
        # polls this endpoint while a download is in flight; see
        # downloadJournalData() in journal_quality.html.
        "download_progress": get_download_state(),
    }


# ---------------------------------------------------------------------------
# Bulk download (dashboard "Download Data" button)
# ---------------------------------------------------------------------------


def download_journal_data(force: bool = False) -> tuple[bool, str]:
    """Fetch every registered data source into the user data directory.

    Iterates `ALL_SOURCES` in order. Sources marked `required=True`
    (OpenAlex) abort the batch on failure; `required=False` sources are
    best-effort and continue on error.

    Args:
        force: Re-fetch even if data exists and is current version.

    Returns:
        (success, message) tuple. Message format is
        `"Fetched <N1> <label1> + <N2> <label2> + ... in <S>s"` so the
        existing test substring assertions ("100 OpenAlex", "50 DOAJ")
        continue to match.
    """
    data_dir = _get_data_dir()
    sentinel = data_dir / _SENTINEL

    if not force:
        status = get_journal_data_status()
        if status["available"] and not status["needs_update"]:
            # No fresh fetch ran, so invalidate any counts cached from a
            # previous in-process download. Callers keying a "what just
            # happened" summary off `counts` must see None here.
            _download_state["counts"] = None
            return True, "Journal data is already up to date"

    # Disk-space pre-check. The five data sources uncompress to ~1 GB
    # intermediate, plus the compiled DB. Fail fast with a clear message
    # rather than crashing mid-download and leaving a corrupt tmp file.
    import shutil as _shutil

    from ..constants import JOURNAL_QUALITY_MIN_FREE_DISK_BYTES

    try:
        free_bytes = _shutil.disk_usage(str(data_dir)).free
    except OSError:
        logger.warning(
            f"Could not check free disk space for {data_dir}; proceeding."
        )
        free_bytes = None
    if (
        free_bytes is not None
        and free_bytes < JOURNAL_QUALITY_MIN_FREE_DISK_BYTES
    ):
        # No fetch ran → invalidate any stale counts from a prior call
        # so `get_download_state()["counts"]` cannot leak them.
        _download_state["counts"] = None
        return False, (
            f"Insufficient disk space: "
            f"{free_bytes / (1024**3):.1f} GB available, "
            f"{JOURNAL_QUALITY_MIN_FREE_DISK_BYTES / (1024**3):.0f} GB required."
        )

    # Atomic sentinel creation (O_CREAT | O_EXCL). Replaces the previous
    # exists()+touch() TOCTOU race so two concurrent download triggers
    # (dashboard click + scheduler) cannot both proceed.
    #
    # Stale-sentinel recovery. Two triggers are checked each call:
    #
    #   1. PID-based: the sentinel holds the PID of the process that
    #      created it. If that PID is not alive, the owner process
    #      crashed or was killed and the sentinel is orphan.
    #   2. Age-based (fallback): if the sentinel is older than
    #      _SENTINEL_STALE_SECS we reclaim it even if the PID check is
    #      inconclusive (exotic environments, unreadable sentinel).
    #
    # The startup hook in _clear_orphan_sentinel_on_startup handles the
    # common case of "server was restarted mid-download"; these two
    # runtime checks cover the case where the server is still running
    # but the download worker thread itself crashed out of the sentinel.
    import os

    def _sentinel_owner_alive() -> bool:
        try:
            owner_pid = int(sentinel.read_text(encoding="utf-8").strip())
        except (OSError, ValueError):
            return False  # unreadable / malformed → treat as orphan
        if owner_pid == os.getpid():
            # Same process. Something is wrong (we should never race
            # with ourselves — the module-level lock guards that) but
            # err on the side of "alive" to avoid self-nuking.
            return True
        try:
            os.kill(owner_pid, 0)  # signal 0 = liveness probe
        except ProcessLookupError:
            return False
        except PermissionError:
            # Process exists, owned by another user — treat as alive.
            return True
        except OSError:
            return False
        return True

    def _try_claim_sentinel() -> bool:
        """Create the sentinel + stamp our PID. Returns True on success."""
        try:
            with sentinel.open("x", encoding="utf-8") as f:
                f.write(str(os.getpid()))
            return True
        except FileExistsError:
            return False

    if not _try_claim_sentinel():
        try:
            age = time.time() - sentinel.stat().st_mtime
        except OSError:
            age = 0
        orphan = not _sentinel_owner_alive()
        if orphan or age > _SENTINEL_STALE_SECS:
            reason = (
                "owner process not alive"
                if orphan
                else f"age {age:.0f}s > {_SENTINEL_STALE_SECS}s"
            )
            logger.warning(
                f"Reclaiming stale .downloading sentinel ({reason}); "
                "previous download likely crashed without cleanup."
            )
            sentinel.unlink(missing_ok=True)
            if not _try_claim_sentinel():
                # Lost a race with another caller reclaiming the same
                # stale sentinel — bow out cleanly.
                _download_state["counts"] = None
                return False, "Download already in progress"
        else:
            _download_state["counts"] = None
            return False, "Download already in progress"
    try:
        start = time.time()
        counts: dict[str, int] = {}
        parts: list[str] = []

        # Reset per-source state to a clean "pending" row for each
        # known source. The dashboard renders one row per entry.
        _download_state["state"] = "running"
        _download_state["started_at"] = start
        _download_state["finished_at"] = None
        _download_state["error_msg"] = None
        # Invalidate counts from any prior run — callers inspecting the
        # structured summary must not see stale data if this download
        # fails or is still running.
        _download_state["counts"] = None
        _download_state["db_build"] = {"state": "pending", "detail": ""}
        _download_state["sources"] = {
            src.key: {
                "name": src.name,
                "state": "pending",
                "detail": "",
                "percent": 0,
                "count": 0,
                "required": src.required,
            }
            for src in ALL_SOURCES
        }

        def _fetch_one(src):
            """Worker: run one source's fetch, mirror state as it goes."""
            _set_source_state(
                src.key, state="running", detail="downloading", percent=5
            )

            # Per-partition callback: the chunked sources (openalex
            # sources + institutions) call this on every partition so
            # the dashboard's bar moves smoothly. One-shot sources
            # don't call it and stay at the initial 5% → final 100%.
            def _on_progress(done, total, detail):
                pct = int(5 + (done / total) * 90) if total > 0 else 5
                _set_source_state(
                    src.key,
                    state="running",
                    detail=detail,
                    percent=max(5, min(95, pct)),
                )

            try:
                shim_name = _FETCH_SHIM_NAMES.get(src.key)
                if shim_name:
                    # bearer:disable python_lang_code_injection
                    # _FETCH_SHIM_NAMES is a hardcoded dict (line 188); the
                    # late-bound globals() lookup is needed for mock.patch
                    # compatibility in tests. No user input reaches the key.
                    n = globals()[shim_name](data_dir, progress_cb=_on_progress)
                else:
                    n = src.fetch(data_dir, progress_cb=_on_progress)
                _set_source_state(
                    src.key,
                    state="success",
                    detail=f"{n} {src.count_label}",
                    percent=100,
                    count=n,
                )
                return (src, n, None)
            except Exception as exc:
                logger.exception(
                    f"{src.name} fetch failed "
                    f"({'required' if src.required else 'optional'})"
                )
                _set_source_state(
                    src.key,
                    state="error",
                    detail=exc.__class__.__name__,
                    percent=100,
                )
                return (src, 0, exc)

        # Parallel fetch — every source streams from a different host
        # (openalex S3, DOAJ CSV, raw.githubusercontent.com, api.openalex.org
        # for institutions). No single-host contention; the wall-clock is
        # dominated by the slowest source (OpenAlex snapshot ~30-60 s).
        from concurrent.futures import ThreadPoolExecutor

        with ThreadPoolExecutor(
            max_workers=max(1, len(ALL_SOURCES)),
            thread_name_prefix="journal-dl",
        ) as pool:
            results = list(pool.map(_fetch_one, ALL_SOURCES))

        required_failure = None
        for src, n, exc in results:
            counts[src.key] = n
            parts.append(f"{n} {src.count_label}")
            if exc is not None and src.required:
                required_failure = (src, exc)

        if required_failure is not None:
            src, _exc = required_failure
            msg = f"Failed to fetch {src.name}. Check your network connection."
            _download_state["state"] = "error"
            _download_state["error_msg"] = msg
            _download_state["finished_at"] = time.time()
            return False, msg

        # Write version marker. Per-source key names are preserved for
        # any external consumer that might read them, even though no
        # production code does today.
        version_file = data_dir / "version.json"
        with open(version_file, "w", encoding="utf-8") as f:
            json.dump(
                {
                    "version": JOURNAL_DATA_VERSION,
                    "downloaded_at": time.strftime("%Y-%m-%dT%H:%M:%SZ"),
                    "openalex_count": counts.get("openalex", 0),
                    "doaj_count": counts.get("doaj", 0),
                    "jabref_count": counts.get("jabref", 0),
                    "predatory_count": counts.get("predatory", 0),
                },
                f,
            )

        # Rebuild journal_quality.db synchronously from the freshly
        # downloaded gz files. The build is the ONLY writer of this
        # file; everything else opens it read-only via mode=ro.
        # Also clean up any leftover legacy journal_reference.db files
        # from before the rename so existing installs don't carry junk.
        legacy_db = data_dir / "journal_reference.db"
        if legacy_db.exists():
            try:
                import os as _os

                # bearer:disable python_lang_file_permissions
                _os.chmod(legacy_db, 0o644)
                legacy_db.unlink()
                logger.info(
                    "Removed legacy journal_reference.db "
                    "(replaced by journal_quality.db)"
                )
            except OSError:
                logger.exception("Could not remove legacy DB")

        _download_state["db_build"] = {
            "state": "running",
            "detail": "parsing bundled data",
        }
        db_build_error: Optional[str] = None
        try:
            from .db import DB_FILENAME, build_db

            new_db_file = data_dir / DB_FILENAME
            if new_db_file.exists():
                import os as _os

                # bearer:disable python_lang_file_permissions
                _os.chmod(new_db_file, 0o644)
                new_db_file.unlink()
            build_db(data_dir=data_dir, output_path=new_db_file)
        except Exception as exc:
            logger.exception(
                "Failed to rebuild journal_quality.db; "
                "the runtime accessor will lazy-build on next access"
            )
            # SchemaDriftError messages are developer-authored literals
            # (no SQL, paths, or stack fragments) so they're safe to
            # surface — operators need to see *which* field drifted to
            # act on it. For any other exception, fall back to the
            # class name only, per CodeQL "Information exposure through
            # an exception" (alerts 7650, 7684). The full trace always
            # stays in logger.exception above (server-side only).
            from .data_sources.openalex import SchemaDriftError

            if isinstance(exc, SchemaDriftError):
                db_build_error = str(exc)
            else:
                db_build_error = exc.__class__.__name__

        elapsed = time.time() - start
        if db_build_error:
            msg = (
                f"Downloaded data ({' + '.join(parts)}) in {elapsed:.0f}s "
                f"but DB build failed ({db_build_error}). "
                f"Lazy-build will retry on next access."
            )
            _download_state["db_build"] = {
                "state": "error",
                "detail": db_build_error,
            }
            _download_state["state"] = "error"
            _download_state["error_msg"] = msg
            _download_state["finished_at"] = time.time()
            return False, msg

        success_msg = f"Fetched {' + '.join(parts)} in {elapsed:.0f}s"
        _download_state["db_build"] = {
            "state": "success",
            "detail": "ready",
        }
        _download_state["state"] = "success"
        _download_state["error_msg"] = None
        _download_state["finished_at"] = time.time()
        # Publish structured counts for callers that want to render a
        # user-facing summary without echoing `success_msg`. All values
        # are ints populated from source `.fetch()` returns above.
        _download_state["counts"] = dict(counts)
        return True, success_msg

    finally:
        sentinel.unlink(missing_ok=True)


_ensure_cache: Optional[tuple[float, tuple[Optional[Path], bool]]] = None
_ENSURE_CACHE_TTL = 30.0  # seconds


def ensure_journal_data(
    auto_download: bool = True,
) -> tuple[Optional[Path], bool]:
    """Ensure journal data is available, optionally triggering a bulk fetch.

    Returns:
        (data_dir, is_available) — data_dir is None if unavailable.

    Thundering-herd guard: when a search runs, every search engine's
    reputation-filter worker (~30 threads) calls this concurrently.
    Without the cache, 29 of them race to create the sentinel and
    each logs a WARNING. One call does the real work; the rest get
    the cached answer for 30 seconds. The success path (data files
    present) is already fast — we only cache the negative / race
    result, which is the noisy path.
    """
    global _ensure_cache

    user_dir = _get_data_dir()
    if (user_dir / "openalex_sources.json.gz").exists():
        # Positive path is cheap (one stat call) — no need to cache.
        return user_dir, True

    now = time.time()
    if _ensure_cache is not None:
        ts, cached = _ensure_cache
        if now - ts < _ENSURE_CACHE_TTL:
            return cached

    if auto_download:
        logger.info(
            "Journal data not found — fetching from upstream sources..."
        )
        success, message = download_journal_data()
        if success:
            logger.info(message)
            _ensure_cache = (now, (user_dir, True))
            return user_dir, True
        logger.warning(f"Journal data fetch failed: {message}")

    result: tuple[Optional[Path], bool] = (None, False)
    _ensure_cache = (now, result)
    return result
