"""Read-only SQLAlchemy accessor for the compiled journal-quality DB.

INVARIANT: this module's `JournalQualityDB` class **never writes**.
The runtime engine is opened with SQLite URI flags `mode=ro` and
`immutable=1`, the file is `chmod 0o444` after every build, and a
pre-commit hook bans cross-module opens of the file without `mode=ro`.

The only writer is `build_db()` in this same module, which opens its
own short-lived writable engine, populates the schema, runs ANALYZE
+ VACUUM, closes the engine, and chmods the file back to 0o444.

The DB compiles five gzipped JSON snapshots (downloaded by
`journal_quality.downloader`) into one queryable file:

- OpenAlex sources → `sources` table (with predatory + DOAJ flags)
- Stop Predatory Journals → `predatory_journals/_publishers/_hijacked`
- DOAJ → cross-referenced into `sources`
- JabRef abbreviations → `abbreviations` table
- OpenAlex Institutions → `institutions` table

Built fresh on every download, no migrations.
"""

from __future__ import annotations

import gzip
import json
import os
import secrets
import sqlite3
import sys
import threading
import time
from contextlib import contextmanager
from pathlib import Path
from typing import Iterable, Iterator, Optional

from loguru import logger
from sqlalchemy import create_engine, func, or_, select
from sqlalchemy.engine import Engine
from sqlalchemy.exc import DatabaseError, OperationalError
from sqlalchemy.orm import Session, sessionmaker

from .models import (
    Abbreviation,
    Institution,
    JournalQualityBase,
    PredatoryHijacked,
    PredatoryJournal,
    PredatoryPublisher,
    Source,
)
from ..constants import PREDATORY_WHITELIST_HINDEX
from ..utilities.citation_normalizer import normalize_issn
from .scoring import (
    derive_quality_score,
    institution_score_from_h_index,
    normalize_name,
)

DB_FILENAME = "journal_quality.db"
_BATCH_SIZE = 5000

# Bump when the reference DB schema (models.py) changes in a way that
# requires a rebuild even if the upstream data version hasn't changed.
# Stamped as SQLite `PRAGMA user_version` during build_db; checked on
# _ensure_engine. Separate from JOURNAL_DATA_VERSION (downloader.py)
# which tracks upstream source-data freshness.
JOURNAL_QUALITY_SCHEMA_VERSION = 3

# Quality tier → score range, used by the dashboard tier filter.
_TIER_RANGES = {
    "elite": (9, 10),
    "strong": (7, 8),
    "moderate": (5, 6),
    "low": (3, 4),
    "predatory": (1, 2),
}

# Columns safe to use in ORDER BY (prevents injection via the dashboard
# `sort` query parameter).
_SORT_COLUMNS = frozenset(
    {
        "name",
        "quality",
        "quartile",
        "h_index",
        "impact_factor",
        "score_source",
        "source_type",
        "publisher",
        "is_predatory",
    }
)

# Max length for user-supplied search strings. Even with LIKE
# wildcards escaped, a 10 KB pattern against 217K rows is slow enough
# to matter for CPU budget under concurrent requests.
_MAX_SEARCH_LEN = 100


def _escape_like(s: str) -> str:
    """Escape SQL LIKE metacharacters so user-supplied search strings
    can't force degenerate full-table scans via ``%`` or ``_``.

    Pair with ``.like(pattern, escape="/")`` at the call site.
    """
    return s.replace("/", "//").replace("%", "/%").replace("_", "/_")


# ---------------------------------------------------------------------------
# Read-only accessor
# ---------------------------------------------------------------------------


class JournalQualityDB:
    """Read-only SQLAlchemy 2.0 accessor for `journal_quality.db`.

    All filter hot-path methods return plain dicts (not mapped Source
    objects) so call sites in `journal_reputation_filter.py` keep the
    same call shape they had against the dict-based predecessor. The
    dashboard methods can return either dicts or Source instances —
    they're called once per page view so the ORM overhead is fine.
    """

    def __init__(self) -> None:
        self._engine: Optional[Engine] = None
        self._SessionLocal: Optional[sessionmaker[Session]] = None
        # RLock (not Lock): _ensure_engine holds the lock while calling
        # _build_or_raise → build_db → reset_db → self.reset(), which
        # re-acquires the same lock. A non-reentrant Lock would deadlock
        # the very first request on a fresh install.
        self._lock = threading.RLock()
        # Whether we've already logged a stale-data-version warning for
        # this engine lifetime. Prevents log spam — one WARNING per
        # server start is enough to surface the problem to admins.
        self._stale_version_warned = False

    # --- engine + session lifecycle ---

    def _resolve_db_path(self) -> Path:
        from ..config.paths import get_journal_data_directory

        return get_journal_data_directory() / DB_FILENAME

    def _ensure_engine(self) -> None:
        # Acquire lock BEFORE first read to avoid DCLP publication hazard.
        # With the GIL this is safe on CPython but explicit locking makes
        # the happens-before relationship clear and portable.
        with self._lock:
            if self._engine is not None:
                return
            path = self._resolve_db_path()
            if not path.exists():
                self._build_or_raise(path)
            else:
                # Validate existing file before wiring up the read-only
                # engine. Catches two failure modes at open time instead
                # of letting them propagate to first query:
                #   1. Schema drift — ORM changed since this file was
                #      built (PRAGMA user_version mismatch) → rebuild.
                #   2. Corruption — file exists but isn't a valid DB
                #      (truncated build, disk error) → rebuild.
                if not self._validate_existing_db(path):
                    self._build_or_raise(path)

            # mode=ro + immutable=1: SQLite physically refuses writes,
            # skips locking entirely, and reads via mmap. The OS page
            # cache holds one shared resident copy of the hot pages.
            #
            # Use a creator callback because SQLAlchemy's URL parser
            # eats the ?mode=ro&immutable=1 query string before it can
            # reach sqlite3. The creator builds the connection directly
            # with the SQLite URI flags intact.
            def _make_ro_conn() -> sqlite3.Connection:
                return sqlite3.connect(
                    f"file:{path}?mode=ro&immutable=1",
                    uri=True,
                    check_same_thread=False,
                )

            # StaticPool: with immutable=1 SQLite skips locking and the
            # OS page cache handles concurrency. A single shared connection
            # is safe and avoids the default QueuePool's 15-connection
            # footprint that offers no benefit for immutable reads.
            from sqlalchemy.pool import StaticPool

            engine = create_engine(
                "sqlite://",
                creator=_make_ro_conn,
                poolclass=StaticPool,
                echo=False,
            )
            session_local = sessionmaker(bind=engine, expire_on_commit=False)
            # Publish both together so readers never see engine-without-session.
            self._engine = engine
            self._SessionLocal = session_local
            logger.info(f"Opened journal_quality.db (read-only): {path}")
            # One-shot check: is the DATA version (the one the sources
            # JSON + build logic produce) behind the bundled latest?
            # Schema drift is already handled by `_validate_existing_db`
            # via ``PRAGMA user_version``. A data-version mismatch is a
            # different concern: the DB schema is fine, but the scoring
            # logic (e.g. the repository cap) or source snapshots have
            # been updated since this file was built. The hot path
            # (filter scoring) would silently serve stale scores if we
            # didn't surface the mismatch anywhere but the admin
            # dashboard. Log once, don't auto-rebuild — user consent
            # via the dashboard "Download Data" button remains the
            # explicit refresh trigger.
            self._warn_on_stale_data_version(path.parent)

    def _warn_on_stale_data_version(self, data_dir: Path) -> None:
        """Log WARNING once if ``version.json`` is behind ``JOURNAL_DATA_VERSION``."""
        if self._stale_version_warned:
            return
        # Lazy import to avoid any downloader → db cycle even though
        # today's module graph doesn't have one.
        from .downloader import JOURNAL_DATA_VERSION

        version_file = data_dir / "version.json"
        if not version_file.exists():
            return  # Brand-new install — the dashboard's banner handles this.
        try:
            with open(version_file) as f:
                info = json.load(f)
            installed = info.get("version")
        except (json.JSONDecodeError, OSError):
            return  # Malformed — dashboard's banner surfaces; don't double-log.
        if installed and installed != JOURNAL_DATA_VERSION:
            logger.warning(
                f"journal_quality data version is stale: on-disk={installed!r} "
                f"bundled-latest={JOURNAL_DATA_VERSION!r}. "
                f"Scoring is continuing with the older data. Visit "
                f"/metrics/journals and click 'Download Data' to refresh."
            )
            self._stale_version_warned = True

    def _validate_existing_db(self, path: Path) -> bool:
        """Return True if the existing DB file is usable as-is.

        A version of 0 means the file was built before schema stamping
        existed and is grandfathered in — we don't force a rebuild just
        because the stamp is missing. A non-zero version that doesn't
        match the current schema is a real drift signal and triggers a
        rebuild. File-open errors also trigger a rebuild.
        """
        from ..utilities.resource_utils import safe_close

        conn: Optional[sqlite3.Connection] = None
        try:
            conn = sqlite3.connect(f"file:{path}?mode=ro", uri=True)
            version = conn.execute("PRAGMA user_version").fetchone()[0]
            if version != 0 and version != JOURNAL_QUALITY_SCHEMA_VERSION:
                logger.warning(
                    f"journal_quality.db schema_version={version}, "
                    f"expected {JOURNAL_QUALITY_SCHEMA_VERSION} — "
                    f"rebuilding"
                )
                # NB: no explicit safe_close here — the finally block
                # handles closing. Calling it twice produced a spurious
                # "Cannot operate on a closed database" warning on
                # every schema-triggered rebuild.
                self._unlink_unusable_db(path)
                return False
            # Cheap sanity check — confirms the file is a valid DB.
            conn.execute("SELECT 1 FROM sqlite_master LIMIT 1").fetchone()
            return True
        except (sqlite3.DatabaseError, OSError):
            logger.exception(
                f"journal_quality.db at {path} is unusable; rebuilding"
            )
            self._unlink_unusable_db(path)
            return False
        finally:
            if conn is not None:
                safe_close(conn, "journal_quality validate")

    @staticmethod
    def _unlink_unusable_db(path: Path) -> None:
        """Best-effort cleanup of a corrupted / schema-drifted DB file.

        Corruption was already logged by the caller (``_validate_existing_db``).
        Both operations below are best-effort — on failure we *log and
        continue* rather than raise, because the build path will rebuild
        the file regardless. But we don't silence: if chmod / unlink
        fails (permissions, read-only mount, file held open on Windows)
        the next build will likely also fail and the user needs the
        warning to diagnose the real problem.
        """
        try:
            # bearer:disable python_lang_file_permissions
            os.chmod(path, 0o644)
        except OSError:
            logger.warning(
                f"Could not chmod 0644 on unusable DB {path} before "
                f"unlink (continuing to unlink attempt)"
            )
        try:
            path.unlink()
        except OSError:
            logger.warning(
                f"Could not unlink unusable DB {path} (will be "
                f"overwritten on next build)"
            )

    def _build_or_raise(self, path: Path) -> None:
        """Lazy-build the DB on first access if it's missing."""
        from .downloader import ensure_journal_data

        data_dir, available = ensure_journal_data()
        if not available:
            raise FileNotFoundError(
                "Journal data files not available. "
                "Check your network connection or download manually "
                "from the dashboard."
            )
        logger.info(f"Building {DB_FILENAME} from data files...")
        build_db(data_dir=data_dir, output_path=path)

    @contextmanager
    def session(self) -> Iterator[Session]:
        """Yield a read-only SQLAlchemy session.

        If the underlying file becomes corrupt mid-session (e.g. a
        rebuild ran and this engine is pointed at a now-unlinked inode),
        DatabaseError propagates but we drop the cached engine so the
        next call rebuilds cleanly instead of failing forever.
        """
        self._ensure_engine()
        if self._SessionLocal is None:
            raise RuntimeError("JournalQualityDB engine failed to initialize")
        from ..utilities.resource_utils import safe_close

        s = self._SessionLocal()
        try:
            yield s
        except (OperationalError, DatabaseError):
            logger.exception("journal_quality.db error — resetting engine")
            safe_close(s, "journal_quality session")
            self.reset()
            raise
        else:
            safe_close(s, "journal_quality session")

    @property
    def available(self) -> bool:
        try:
            self._ensure_engine()
            return True
        except FileNotFoundError:
            return False

    def reset(self) -> None:
        """Drop the cached engine — call after `build_db` rebuilds the file."""
        with self._lock:
            if self._engine is not None:
                self._engine.dispose()
                self._engine = None
                self._SessionLocal = None
                logger.info("Reset journal_quality.db engine")

    # --- filter hot path: return plain dicts ---

    def lookup_openalex(
        self,
        *,
        source_id: Optional[str] = None,
        issn: Optional[str] = None,
        name: Optional[str] = None,
    ) -> Optional[dict]:
        """Look up a source by OpenAlex ID, ISSN, or name.

        Returns a dict with the same shape the dict-based predecessor
        produced (`name`, `type`, `h_index`, `impact_factor`,
        `is_in_doaj`, `publisher`, `issn_l`) so the filter code at
        `journal_reputation_filter.py` doesn't need to change.
        """
        issn = normalize_issn(issn)
        try:
            self._ensure_engine()
        except FileNotFoundError:
            return None
        with self.session() as s:
            row = self._lookup_source_row(s, source_id, issn, name)
            return _source_to_lookup_dict(row) if row else None

    # Alias used by the dashboard / future call sites
    lookup_source = lookup_openalex

    def count_predatory_by_names(self, names: Iterable[str]) -> int:
        """Count how many of the given journal names are flagged predatory.

        One SQL round-trip using ``WHERE name_lower IN (…) AND is_predatory = TRUE``.
        Names are normalized (NFKC + lower + strip) so the caller can pass raw
        display names; matches the normalization used at build time.

        Used by the per-user metrics dashboard to report a global "N predatory
        journals across all your research" stat without making N round trips
        to the reference DB. Returns 0 if the reference DB is missing or if
        ``names`` is empty.

        .. note::
           Deliberately unchunked. SQLite's ``SQLITE_MAX_VARIABLE_NUMBER``
           has been 250,000 since SQLite 3.32 (2020); Python 3.11 ships
           with 3.45.1. A heavy user with 100k distinct container_titles
           is still well under the limit. Re-confirmed in the PR #3081
           audit — no chunking needed.
        """
        normed = {normalize_name(n) for n in names if n}
        normed.discard("")
        if not normed:
            return 0
        try:
            self._ensure_engine()
        except FileNotFoundError:
            return 0
        with self.session() as s:
            stmt = select(func.count(Source.id)).where(
                Source.name_lower.in_(normed),
                Source.is_predatory.is_(True),
            )
            result = s.execute(stmt).scalar()
            return int(result or 0)

    def lookup_sources_batch(self, names: Iterable[str]) -> dict:
        """Batch-look-up multiple journal names in one query.

        Takes an iterable of raw display names and returns a
        ``{normalized_name: dashboard_dict}`` map for every name that
        matched a Source. Names that didn't match are simply absent
        from the result (caller decides how to handle misses).

        Dashboard hot path: the ``/api/journals/user-research`` endpoint
        collects up to 200 unique ``container_title`` values from the
        user's Papers and hands them in here — one SQL round-trip vs.
        200 per-row lookups.

        Normalization matches ``normalize_name`` (NFKC + lower + strip)
        so the reference DB's ``name_lower`` column hits directly. No
        "the " / "proceedings of" fallback tiers — those live in
        ``_lookup_source_row`` for precision per-call; the batch path
        is for dashboard display where a miss is acceptable.

        Chunked at 900 params per chunk. Defensive: SQLite's actual
        limit (``SQLITE_MAX_VARIABLE_NUMBER``) has been 250,000 since
        3.32 (2020) — we could easily put the whole batch in one IN —
        but 900 keeps us well under any older embedded-SQLite ceiling
        a deployment might pin to.
        """
        normed = [normalize_name(n) for n in names if n]
        normed = [n for n in normed if n]
        if not normed:
            return {}
        try:
            self._ensure_engine()
        except FileNotFoundError:
            return {}
        # De-duplicate while preserving insertion order for stable
        # iteration in tests.
        seen: set = set()
        uniq: list = []
        for n in normed:
            if n not in seen:
                seen.add(n)
                uniq.append(n)

        out: dict = {}
        CHUNK = 900
        with self.session() as s:
            for i in range(0, len(uniq), CHUNK):
                batch = uniq[i : i + CHUNK]
                stmt = select(Source).where(Source.name_lower.in_(batch))
                for row in s.scalars(stmt):
                    out[row.name_lower] = _source_to_dashboard_dict(row)
        return out

    def lookup_doaj(self, *, issn: Optional[str] = None) -> Optional[dict]:
        issn = normalize_issn(issn)
        if not issn:
            return None
        try:
            self._ensure_engine()
        except FileNotFoundError:
            return None
        with self.session() as s:
            stmt = (
                select(Source)
                .where(Source.issn == issn, Source.is_in_doaj.is_(True))
                .limit(1)
            )
            row = s.scalars(stmt).first()
            if row is None:
                return None
            return {
                "name": row.name,
                "has_seal": row.has_doaj_seal,
                "publisher": row.publisher,
            }

    def is_in_doaj(self, issn: Optional[str]) -> bool:
        return self.lookup_doaj(issn=issn) is not None

    def has_doaj_seal(self, issn: Optional[str]) -> bool:
        entry = self.lookup_doaj(issn=issn)
        return bool(entry and entry.get("has_seal"))

    def is_predatory(
        self,
        *,
        journal_name: Optional[str] = None,
        publisher_name: Optional[str] = None,
    ) -> tuple[bool, Optional[str]]:
        """Check if a journal/publisher is on the predatory list.

        Looks up the dedicated predatory tables (NOT just the
        `is_predatory` flag on `Source`), so checks work for arbitrary
        input names that aren't in OpenAlex.
        """
        try:
            self._ensure_engine()
        except FileNotFoundError:
            return False, None
        with self.session() as s:
            if journal_name:
                norm = normalize_name(journal_name)
                if s.get(PredatoryJournal, norm) is not None:
                    return True, "stop-predatory-journals"
                if s.get(PredatoryHijacked, norm) is not None:
                    return True, "stop-predatory-hijacked"

            if publisher_name:
                pub_norm = normalize_name(publisher_name)
                if s.get(PredatoryPublisher, pub_norm) is not None:
                    return True, "stop-predatory-publishers"
                # Substring scan over long entries (~1162 rows)
                stmt = select(PredatoryPublisher.name_lower).where(
                    PredatoryPublisher.is_long.is_(True)
                )
                for (entry,) in s.execute(stmt).all():
                    if pub_norm in entry or entry in pub_norm:
                        return True, "stop-predatory-publishers"

        return False, None

    def is_whitelisted(
        self,
        *,
        issn: Optional[str] = None,
        name: Optional[str] = None,
    ) -> bool:
        if self.is_in_doaj(issn):
            return True
        oa = self.lookup_openalex(issn=issn, name=name)
        if oa and (oa.get("h_index") or 0) > PREDATORY_WHITELIST_HINDEX:
            return True
        return False

    def lookup_institution(
        self,
        *,
        ror_id: Optional[str] = None,
        openalex_id: Optional[str] = None,
        name: Optional[str] = None,
    ) -> Optional[dict]:
        """Look up an institution.

        Order: openalex_id → ror → name. Returns a dict with full-name
        keys (``name``, ``country``, ``type``, ``h_index``,
        ``impact_factor``, ``works_count``, ``cited_by_count``, ``ror_id``)
        or ``None`` if no match. The on-disk snapshot uses one-character
        keys (``n``, ``c``, etc.) for space efficiency; the accessor
        returns full names instead for legibility and schema robustness.
        """
        try:
            self._ensure_engine()
        except FileNotFoundError:
            return None
        with self.session() as s:
            row: Optional[Institution] = None

            if openalex_id:
                sid = openalex_id.split("/")[-1]
                row = s.get(Institution, sid)

            if row is None and ror_id:
                ror = ror_id.rstrip("/").split("/")[-1]
                stmt = (
                    select(Institution)
                    .where(Institution.ror_id == ror)
                    .limit(1)
                )
                row = s.scalars(stmt).first()

            if row is None and name:
                norm = normalize_name(name)
                stmt = (
                    select(Institution)
                    .where(Institution.name_lower == norm)
                    .limit(1)
                )
                row = s.scalars(stmt).first()

            return _institution_to_dict(row) if row else None

    def score_from_affiliations(self, affiliations: list) -> Optional[int]:
        """Derive a score from author affiliations in ONE SQL query."""
        if not affiliations:
            return None

        openalex_ids: list[str] = []
        ror_ids: list[str] = []
        names: list[str] = []

        for aff in affiliations:
            if isinstance(aff, str):
                names.append(normalize_name(aff))
            elif isinstance(aff, dict):
                if oid := (aff.get("openalex_id") or aff.get("id")):
                    openalex_ids.append(oid.split("/")[-1])
                if rid := aff.get("ror"):
                    ror_ids.append(rid.rstrip("/").split("/")[-1])
                if nm := aff.get("name"):
                    names.append(normalize_name(nm))

        if not (openalex_ids or ror_ids or names):
            return None

        try:
            self._ensure_engine()
        except FileNotFoundError:
            return None

        clauses = []
        if openalex_ids:
            clauses.append(Institution.openalex_id.in_(openalex_ids))
        if ror_ids:
            clauses.append(Institution.ror_id.in_(ror_ids))
        if names:
            clauses.append(Institution.name_lower.in_(names))

        with self.session() as s:
            stmt = select(func.max(Institution.h_index)).where(
                or_(*clauses), Institution.h_index.is_not(None)
            )
            best_h = s.scalar(stmt)

        # Single source of truth for institution scoring lives in
        # scoring.py — delegate so the build phase, the runtime filter,
        # and this affiliation-salvage path can never disagree.
        return institution_score_from_h_index(best_h)

    # Static passthrough so the filter can call dm.derive_quality_score(...)
    # without importing from .scoring directly. Single home for the
    # scoring rules in scoring.py.
    derive_quality_score = staticmethod(derive_quality_score)

    def expand_abbreviation(self, name: str) -> Optional[str]:
        if not name:
            return None
        try:
            self._ensure_engine()
        except FileNotFoundError:
            return None
        normalized = normalize_name(name)
        with self.session() as s:
            row = s.get(Abbreviation, normalized)
            if row is not None:
                return row.full_name
            no_dots = normalized.replace(".", "").strip()
            if no_dots != normalized:
                row = s.get(Abbreviation, no_dots)
                if row is not None:
                    return row.full_name
        return None

    # --- internal source lookup with name fallbacks ---

    def _lookup_source_row(
        self,
        s: Session,
        source_id: Optional[str],
        issn: Optional[str],
        name: Optional[str],
    ) -> Optional[Source]:
        if source_id:
            sid = source_id.split("/")[-1] if "/" in source_id else source_id
            stmt = (
                select(Source).where(Source.openalex_source_id == sid).limit(1)
            )
            row = s.scalars(stmt).first()
            if row is not None:
                return row

        if issn:
            stmt = select(Source).where(Source.issn == issn).limit(1)
            row = s.scalars(stmt).first()
            if row is not None:
                return row

        if name:
            norm = normalize_name(name)
            row = self._fetch_by_name_lower(s, norm)
            if row is not None:
                return row
            # Try with/without "the " prefix (~5K journals have it)
            if norm.startswith("the "):
                row = self._fetch_by_name_lower(s, norm[4:])
            else:
                row = self._fetch_by_name_lower(s, "the " + norm)
            if row is not None:
                return row
            # Strip "proceedings of (the) (conference on) " prefix
            stripped = norm
            for prefix in (
                "proceedings of the conference on ",
                "proceedings of the ",
                "proceedings of ",
            ):
                if stripped.startswith(prefix):
                    stripped = stripped[len(prefix) :]
                    break
            if stripped != norm:
                row = self._fetch_by_name_lower(s, stripped)
                if row is not None:
                    return row

            # MEDLINE-style "Title : long subtitle" — try the segment
            # before the colon. Catches PubMed names like
            #   "Molecular therapy : the journal of the American Society..."
            # → "Molecular therapy"
            if " : " in norm:
                head = norm.split(" : ", 1)[0].strip()
                if head and head != norm:
                    row = self._fetch_by_name_lower(s, head)
                    if row is not None:
                        return row

            # MEDLINE-style "Title. Section name" — try the segment
            # before the first period. Catches PubMed names like
            #   "Molecular therapy. Methods and clinical development"
            # but only when the head is meaningfully shorter (we don't
            # want to match "Nat" from "Nat. Commun.").
            if "." in norm:
                head = norm.split(".", 1)[0].strip()
                if head and len(head) >= 6 and head != norm:
                    row = self._fetch_by_name_lower(s, head)
                    if row is not None:
                        return row

        return None

    @staticmethod
    def _fetch_by_name_lower(s: Session, name_lower: str) -> Optional[Source]:
        stmt = select(Source).where(Source.name_lower == name_lower).limit(1)
        return s.scalars(stmt).first()

    # --- dashboard queries ---

    def get_summary(self) -> dict:
        if not self.available:
            return {
                "total": 0,
                "avg_quality": 0,
                "avg_h_index": None,
                "predatory_count": 0,
                "doaj_count": 0,
                "seal_count": 0,
                "llm_count": 0,
            }

        with self.session() as s:
            row = s.execute(
                select(
                    func.count().label("total"),
                    func.round(func.avg(Source.quality), 1).label(
                        "avg_quality"
                    ),
                    func.round(func.avg(Source.h_index)).label("avg_h_index"),
                    func.sum(func.iif(Source.is_predatory, 1, 0)).label(
                        "predatory_count"
                    ),
                    func.sum(func.iif(Source.is_in_doaj, 1, 0)).label(
                        "doaj_count"
                    ),
                    func.sum(func.iif(Source.has_doaj_seal, 1, 0)).label(
                        "seal_count"
                    ),
                    func.sum(
                        func.iif(Source.score_source == "llm", 1, 0)
                    ).label("llm_count"),
                )
            ).first()
        return dict(row._mapping) if row else {}

    def get_quality_distribution(self) -> dict[str, int]:
        if not self.available:
            return {}
        with self.session() as s:
            rows = s.execute(
                select(Source.quality, func.count().label("cnt"))
                .where(Source.quality.is_not(None))
                .group_by(Source.quality)
                .order_by(Source.quality)
            ).all()
        return {str(q): c for q, c in rows}

    def get_source_distribution(self) -> dict[str, int]:
        if not self.available:
            return {}
        with self.session() as s:
            rows = s.execute(
                select(
                    func.coalesce(Source.score_source, "unknown").label("src"),
                    func.count().label("cnt"),
                ).group_by(Source.score_source)
            ).all()
        return {row.src: row.cnt for row in rows}

    def get_journals_page(
        self,
        *,
        page: int = 1,
        per_page: int = 50,
        search: str = "",
        tier: str = "",
        score_source: str = "",
        sort: str = "quality",
        order: str = "desc",
    ) -> tuple[list[dict], int]:
        if not self.available:
            return [], 0

        if sort not in _SORT_COLUMNS:
            sort = "quality"
        if order not in ("asc", "desc"):
            order = "desc"

        wheres: list = []
        if search:
            needle = _escape_like(normalize_name(search)[:_MAX_SEARCH_LEN])
            wheres.append(Source.name_lower.like(f"%{needle}%", escape="/"))
        if tier and tier in _TIER_RANGES:
            lo, hi = _TIER_RANGES[tier]
            wheres.append(Source.quality.between(lo, hi))
        if score_source:
            wheres.append(Source.score_source == score_source)

        sort_col = getattr(Source, sort)
        order_clause = (
            sort_col.desc().nulls_last()
            if order == "desc"
            else sort_col.asc().nulls_last()
        )

        offset = (max(1, page) - 1) * per_page

        with self.session() as s:
            total = (
                s.scalar(
                    select(func.count()).select_from(Source).where(*wheres)
                )
                or 0
            )
            rows = s.scalars(
                select(Source)
                .where(*wheres)
                .order_by(order_clause)
                .limit(per_page)
                .offset(offset)
            ).all()

        return [_source_to_dashboard_dict(r) for r in rows], total

    def get_institutions_page(
        self,
        *,
        page: int = 1,
        per_page: int = 50,
        search: str = "",
        sort: str = "h_index",
        order: str = "desc",
    ) -> tuple[list[dict], int]:
        if not self.available:
            return [], 0

        # Defensive allowlist — matches the pattern in get_journals_page.
        # The ternary below is already safe (non-"desc" falls through to
        # .asc()), but the explicit check prevents future refactors from
        # accidentally interpolating a tainted value into SQL.
        if order not in ("asc", "desc"):
            order = "desc"

        wheres = []
        if search:
            needle = _escape_like(normalize_name(search)[:_MAX_SEARCH_LEN])
            wheres.append(
                Institution.name_lower.like(f"%{needle}%", escape="/")
            )

        sort_col = (
            Institution.h_index if sort == "h_index" else Institution.name
        )
        order_clause = (
            sort_col.desc().nulls_last()
            if order == "desc"
            else sort_col.asc().nulls_last()
        )

        offset = (max(1, page) - 1) * per_page

        with self.session() as s:
            total = (
                s.scalar(
                    select(func.count()).select_from(Institution).where(*wheres)
                )
                or 0
            )
            rows = s.scalars(
                select(Institution)
                .where(*wheres)
                .order_by(order_clause)
                .limit(per_page)
                .offset(offset)
            ).all()

        return [_institution_to_dashboard_dict(r) for r in rows], total


# ---------------------------------------------------------------------------
# Dict adapters — keep filter/dashboard call sites unchanged
# ---------------------------------------------------------------------------


def _source_to_lookup_dict(row: Source) -> dict:
    """Convert a Source row to the dict shape `lookup_openalex` produces.

    Includes `openalex_source_id` so dashboard / test code can chain
    a follow-up `lookup_source(source_id=...)` call. Also exposes
    ``quartile`` so the filter can store it on the per-user Journal row
    and feed it into score derivation.
    """
    return {
        "name": row.name,
        "type": row.source_type,
        "h_index": row.h_index,
        "impact_factor": row.impact_factor,
        "is_in_doaj": row.is_in_doaj,
        "publisher": row.publisher,
        "issn_l": row.issn,
        "openalex_source_id": row.openalex_source_id,
        "quartile": row.quartile,
    }


def _source_to_dashboard_dict(row: Source) -> dict:
    return {
        "name": row.name,
        "quality": row.quality,
        "quartile": row.quartile,
        "cited_by_count": row.cited_by_count,
        "h_index": row.h_index,
        "impact_factor": (
            round(row.impact_factor, 2) if row.impact_factor else None
        ),
        "is_in_doaj": bool(row.is_in_doaj),
        "has_doaj_seal": bool(row.has_doaj_seal),
        "is_predatory": bool(row.is_predatory),
        "predatory_source": row.predatory_source,
        "score_source": row.score_source,
        "source_type": row.source_type,
        "publisher": row.publisher,
        "issn": row.issn,
        "openalex_source_id": row.openalex_source_id,
    }


def _institution_to_dict(row: Institution) -> dict:
    """Public accessor shape for `lookup_institution`.

    The on-disk JSON snapshot uses one-character keys (``n``, ``c``,
    ``t``, …) purely for space efficiency — 200k institutions × seven
    long field names adds real bytes. Callers of the accessor don't
    care about on-disk layout, so here we return the full names to
    keep the public API legible and robust to future schema tweaks.
    """
    return {
        "name": row.name,
        "country": row.country,
        "type": row.type,
        "h_index": row.h_index,
        "impact_factor": row.impact_factor,
        "works_count": row.works_count,
        "cited_by_count": row.cited_by_count,
        "ror_id": row.ror_id,
    }


def _institution_to_dashboard_dict(row: Institution) -> dict:
    return {
        "openalex_id": row.openalex_id,
        "name": row.name,
        "ror_id": row.ror_id,
        "country": row.country,
        "type": row.type,
        "h_index": row.h_index,
        "impact_factor": row.impact_factor,
        "works_count": row.works_count,
        "cited_by_count": row.cited_by_count,
    }


# ---------------------------------------------------------------------------
# Module singleton
# ---------------------------------------------------------------------------


_db: Optional[JournalQualityDB] = None
_db_lock = threading.Lock()


def get_db() -> JournalQualityDB:
    """Get or create the singleton `JournalQualityDB`."""
    global _db
    if _db is None:
        with _db_lock:
            if _db is None:
                _db = JournalQualityDB()
    return _db


# Backwards-compat aliases used by metrics_routes.py and a couple of tests
get_journal_reference_db = get_db
JournalReferenceDB = JournalQualityDB


def reset_db() -> None:
    """Reset the cached engine after a build_db rebuild.

    Held under `_db_lock` so a concurrent `get_db()` call can't see a
    half-disposed singleton — without the lock, Thread B could pass
    the `if _db is None` check in `get_db()` while Thread A is still
    inside `_db.reset()`, then call `_ensure_engine()` which short-
    circuits on the still-set `_engine` and hands back a disposed
    pool. The lock makes the read-then-reset pair atomic with respect
    to `get_db()`'s lazy-init path.
    """
    global _db
    with _db_lock:
        if _db is not None:
            _db.reset()


# ---------------------------------------------------------------------------
# Build phase — the ONLY writer
# ---------------------------------------------------------------------------


def build_db(
    data_dir: Optional[Path] = None,
    output_path: Optional[Path] = None,
) -> None:
    """Compile `journal_quality.db` from the gzipped JSON sources.

    Opens a SHORT-LIVED writable engine, creates the schema, populates
    every table from the gz files, runs ANALYZE + VACUUM, closes the
    engine, then `chmod 0o444` the file.
    """
    start = time.time()

    if data_dir is None:
        from ..config.paths import get_journal_data_directory

        data_dir = get_journal_data_directory()
    if output_path is None:
        output_path = data_dir / DB_FILENAME

    logger.info(
        "Building journal quality reference DB (one-time, "
        "~30s, decompresses ~25 MB of bundled data)…"
    )

    # Sweep stale temp files from prior crashed builds so they don't
    # accumulate. Any .tmp-* older than 1h is assumed dead.
    _sweep_stale_tmp_files(output_path.parent, output_path.name)

    # Build into a unique temp path, then os.replace() atomically at
    # the end. A random suffix (not a fixed .tmp) lets concurrent
    # builders write to separate files instead of racing on the same
    # path — os.replace picks a winner atomically and neither corrupts
    # the live file.
    tmp_path = output_path.with_name(
        f"{output_path.name}.tmp-{os.getpid()}-{secrets.token_hex(4)}"
    )

    write_url = f"sqlite:///{tmp_path}"
    engine = create_engine(write_url, connect_args={"check_same_thread": False})

    try:
        # Pragmas for fast bulk insert. `journal_mode = OFF` plus
        # `synchronous = OFF` is deliberately unsafe for general use but
        # correct here because durability is guaranteed by the temp-file
        # + os.replace() pattern around this block: we write to a unique
        # `.tmp-PID-RAND` path, and on any crash mid-build the incomplete
        # temp file is orphaned (and swept by `_sweep_stale_tmp_files()`
        # on the next build). The live file is only ever moved into place
        # by the atomic `os.replace()` at the bottom of this function —
        # it never sees a partial write. Do NOT copy this pragma set
        # elsewhere without the same atomic rename discipline.
        with engine.connect() as conn:
            conn.exec_driver_sql("PRAGMA journal_mode = OFF")
            conn.exec_driver_sql("PRAGMA synchronous = OFF")
            conn.exec_driver_sql("PRAGMA cache_size = -64000")
            conn.exec_driver_sql("PRAGMA page_size = 4096")

        JournalQualityBase.metadata.create_all(engine)

        SessionWrite = sessionmaker(bind=engine)
        with SessionWrite() as session:
            sources = _load_openalex(data_dir)
            doaj_data = _load_doaj(data_dir)
            pred_data = _load_predatory(data_dir)
            institutions = _load_institutions(data_dir)
            abbreviations = _load_abbreviations(data_dir)

            _populate_predatory(session, pred_data)
            _populate_sources(session, sources, doaj_data, pred_data)
            _populate_institutions(session, institutions)
            _populate_abbreviations(session, abbreviations)
            session.commit()

        with engine.connect() as conn:
            conn.exec_driver_sql("ANALYZE")
            conn.exec_driver_sql("VACUUM")
            # Stamp schema version so _ensure_engine can detect drift
            # without depending on the external version.json.
            conn.exec_driver_sql(
                f"PRAGMA user_version = {JOURNAL_QUALITY_SCHEMA_VERSION}"
            )
    except Exception:
        engine.dispose()
        if tmp_path.exists():
            try:
                # bearer:disable python_lang_file_permissions
                os.chmod(tmp_path, 0o644)
                tmp_path.unlink()
            except OSError:
                logger.exception(f"Failed to clean up tmp DB at {tmp_path}")
        raise

    engine.dispose()

    # Atomically swap tmp into place. os.replace is atomic on POSIX and
    # overwrites an existing output_path if present.
    if output_path.exists():
        # Prior file is chmod 0444 from the previous build — relax it
        # so os.replace can overwrite. Best-effort: if chmod fails
        # (e.g. read-only mount), os.replace will raise and surface
        # the real problem. Log so the cause is visible.
        try:
            # bearer:disable python_lang_file_permissions
            os.chmod(output_path, 0o644)
        except OSError:
            logger.warning(
                f"Could not chmod 0644 on existing {output_path} before "
                f"os.replace; if the replace fails this is likely why"
            )
    os.replace(tmp_path, output_path)

    # OS-level read-only flag — third layer of write protection.
    # POSIX chmod is a no-op on Windows, so we also set the Windows
    # read-only file attribute via SetFileAttributesW. The pre-commit
    # hook check-journal-quality-readonly.py remains the primary
    # defense against accidental writable opens.
    # bearer:disable python_lang_file_permissions
    os.chmod(output_path, 0o444)
    if sys.platform == "win32":
        try:
            import ctypes

            # FILE_ATTRIBUTE_READONLY = 0x1
            ok = ctypes.windll.kernel32.SetFileAttributesW(
                str(output_path), 0x1
            )
            if not ok:
                logger.warning(
                    f"SetFileAttributesW failed on {output_path.name}; "
                    "readonly pre-commit hook is the sole defense."
                )
        except Exception:
            logger.warning(
                f"Could not set Windows readonly attribute on "
                f"{output_path.name}"
            )

    elapsed = time.time() - start
    size_mb = output_path.stat().st_size / (1024 * 1024)
    with sqlite3.connect(
        f"file:{output_path}?mode=ro&immutable=1", uri=True
    ) as _count_conn:
        source_count = _count_conn.execute(
            "SELECT COUNT(*) FROM sources"
        ).fetchone()[0]
    logger.info(
        f"Journal quality DB ready: {source_count} sources, "
        f"{size_mb:.1f} MB in {elapsed:.1f}s ({output_path.name}, chmod 0o444)"
    )

    reset_db()


def _sweep_stale_tmp_files(directory: Path, base_name: str) -> None:
    """Remove journal_quality.db.tmp-* files older than 1h.

    Per-file OSError (vanished between glob+stat, no permission, etc.)
    is logged at debug — the sweep is best-effort and shouldn't stop
    the build, but silent-pass on filesystem errors hides the cause of
    accumulating stale tmp files that would otherwise eat disk over
    time.
    """
    if not directory.exists():
        return
    cutoff = time.time() - 3600
    for tmp in directory.glob(f"{base_name}.tmp-*"):
        try:
            if tmp.stat().st_mtime < cutoff:
                tmp.unlink()
                logger.info(f"Swept stale temp build file: {tmp.name}")
        except OSError:
            logger.debug(f"Could not sweep stale tmp file {tmp.name}")


# ---------------------------------------------------------------------------
# Source-data loaders (used by build_db only)
# ---------------------------------------------------------------------------


def _load_openalex(data_dir: Path) -> dict:
    path = data_dir / "openalex_sources.json.gz"
    if not path.exists():
        raise FileNotFoundError(f"OpenAlex source file not found: {path}")
    with gzip.open(path, "rt", encoding="utf-8") as f:
        data = json.load(f)
    sources = data.get("s", data.get("sources", {}))
    logger.info(f"Loaded {len(sources)} OpenAlex sources")
    return dict(sources)


def _load_doaj(data_dir: Path) -> dict:
    path = data_dir / "doaj_journals.json"
    if not path.exists():
        logger.warning(f"{path} not found — DOAJ cross-ref will be skipped")
        return {}
    with open(path, encoding="utf-8") as f:
        data = json.load(f)
    journals = data.get("journals", {})
    logger.info(f"Loaded {len(journals)} DOAJ entries")
    return dict(journals)


def _load_predatory(data_dir: Path) -> dict:
    """Returns {journals: set, publishers: set, hijacked: set, long_pubs: list}."""
    path = data_dir / "predatory.json"
    if not path.exists():
        logger.warning(f"{path} not found — predatory check will be skipped")
        return {
            "journals": set(),
            "publishers": set(),
            "hijacked": set(),
            "long_pubs": [],
        }

    with open(path, encoding="utf-8") as f:
        data = json.load(f)

    journal_names = {
        normalize_name(e.get("name", ""))
        for e in data.get("journals", [])
        if e.get("name", "").strip()
    }
    publisher_names = {
        normalize_name(e.get("name", ""))
        for e in data.get("publishers", [])
        if e.get("name", "").strip()
    }
    hijacked_names = {
        normalize_name(e.get("hijacked_name", ""))
        for e in data.get("hijacked", [])
        if e.get("hijacked_name", "").strip()
    }
    long_pubs = [
        normalize_name(e.get("name", ""))
        for e in data.get("publishers", [])
        if len(e.get("name", "").strip()) >= 10
    ]
    logger.info(
        f"Loaded predatory: {len(journal_names)} journals, "
        f"{len(publisher_names)} publishers, "
        f"{len(hijacked_names)} hijacked"
    )
    return {
        "journals": journal_names,
        "publishers": publisher_names,
        "hijacked": hijacked_names,
        "long_pubs": long_pubs,
    }


def _load_institutions(data_dir: Path) -> dict:
    path = data_dir / "openalex_institutions.json.gz"
    if not path.exists():
        logger.warning(f"{path} not found — institution tier will be empty")
        return {}
    with gzip.open(path, "rt", encoding="utf-8") as f:
        data = json.load(f)
    institutions = data.get("i", {})
    logger.info(f"Loaded {len(institutions)} institutions")
    return dict(institutions)


def _load_abbreviations(data_dir: Path) -> dict:
    path = data_dir / "jabref_abbreviations.json.gz"
    if not path.exists():
        logger.warning(
            f"{path} not found — abbreviation expansion will be empty"
        )
        return {}
    with gzip.open(path, "rt", encoding="utf-8") as f:
        data = json.load(f)
    mappings = data.get("abbrev_to_full", {})
    logger.info(f"Loaded {len(mappings)} abbreviation mappings")
    return dict(mappings)


# ---------------------------------------------------------------------------
# Table populators
# ---------------------------------------------------------------------------


def _populate_predatory(session: Session, pred: dict) -> None:
    long_set = set(pred.get("long_pubs", []))

    journals = [{"name_lower": n} for n in pred.get("journals", set()) if n]
    if journals:
        session.bulk_insert_mappings(PredatoryJournal, journals)  # type: ignore[arg-type]

    hijacked = [{"name_lower": n} for n in pred.get("hijacked", set()) if n]
    if hijacked:
        session.bulk_insert_mappings(PredatoryHijacked, hijacked)  # type: ignore[arg-type]

    publishers = [
        {"name_lower": n, "is_long": n in long_set}
        for n in pred.get("publishers", set())
        if n
    ]
    if publishers:
        session.bulk_insert_mappings(PredatoryPublisher, publishers)  # type: ignore[arg-type]

    logger.info(
        f"Inserted predatory tables: "
        f"{len(journals)} journals, "
        f"{len(publishers)} publishers, "
        f"{len(hijacked)} hijacked"
    )


def _populate_sources(
    session: Session,
    sources: dict,
    doaj_data: dict,
    pred: dict,
) -> None:
    """Build Source rows with cross-referenced DOAJ + predatory flags."""
    type_map = {"j": "journal", "c": "conference"}
    pred_journals = pred.get("journals", set())
    pred_publishers = pred.get("publishers", set())
    pred_hijacked = pred.get("hijacked", set())

    # Dedup key is (name_lower, issn or "") so journals with separate
    # print and electronic ISSNs in OpenAlex both survive instead of
    # collapsing onto one row.
    seen: dict[tuple[str, str], dict] = {}

    for source_id, compact in sources.items():
        name = (compact.get("n") or "").strip()
        if not name:
            continue

        name_lower = normalize_name(name)
        issn = normalize_issn(compact.get("i"))
        publisher = compact.get("p") or None
        h_index = compact.get("h")
        impact_factor = compact.get("if")
        cited_by_count = compact.get("cb")
        source_type = type_map.get(compact.get("t", ""), compact.get("t", ""))

        doaj_entry = doaj_data.get(issn) if issn else None
        is_in_doaj = doaj_entry is not None
        has_doaj_seal = bool(doaj_entry and doaj_entry.get("has_seal"))

        is_pred = name_lower in pred_journals
        pred_source = "stop-predatory-journals" if is_pred else None
        if not is_pred and publisher:
            pub_norm = normalize_name(publisher)
            if pub_norm in pred_publishers:
                is_pred = True
                pred_source = "stop-predatory-publishers"
        if not is_pred and name_lower in pred_hijacked:
            is_pred = True
            pred_source = "stop-predatory-hijacked"

        # Whitelist override
        if is_pred and (
            is_in_doaj or (h_index or 0) > PREDATORY_WHITELIST_HINDEX
        ):
            is_pred = False
            pred_source = None

        quality = derive_quality_score(
            h_index=h_index,
            is_in_doaj=is_in_doaj,
            has_doaj_seal=has_doaj_seal,
            is_predatory=is_pred,
            source_type=source_type,
        )

        rec = {
            "name": name,
            "name_lower": name_lower,
            "issn": issn,
            "openalex_source_id": source_id,
            "source_type": source_type,
            "publisher": publisher,
            "h_index": h_index,
            "impact_factor": impact_factor,
            "cited_by_count": cited_by_count,
            "quartile": None,  # filled in by the post-pass below
            "quality": quality,
            "is_in_doaj": is_in_doaj,
            "has_doaj_seal": has_doaj_seal,
            "is_predatory": is_pred,
            "predatory_source": pred_source,
            "score_source": "openalex",
        }

        key = (name_lower, issn or "")
        prev = seen.get(key)
        if prev is None or (h_index or 0) > (prev.get("h_index") or 0):
            seen[key] = rec

    # Second pass: DOAJ-only journals (not in OpenAlex). Without this
    # we lose ~4-7K small open-access venues. Keyed by name_lower so
    # we don't double-insert anything OpenAlex already covered.
    openalex_names = {k[0] for k in seen.keys()}
    doaj_added = 0
    for issn, doaj_entry in doaj_data.items():
        name = (doaj_entry.get("name") or "").strip()
        if not name:
            continue
        name_lower = normalize_name(name)
        if name_lower in openalex_names:
            continue
        has_seal = bool(doaj_entry.get("has_seal"))
        publisher = doaj_entry.get("publisher") or None
        quality = derive_quality_score(
            h_index=None,
            is_in_doaj=True,
            has_doaj_seal=has_seal,
            is_predatory=False,
            source_type="journal",
        )
        seen[(name_lower, issn or "")] = {
            "name": name,
            "name_lower": name_lower,
            "issn": issn,
            "openalex_source_id": None,
            "source_type": "journal",
            "publisher": publisher,
            "h_index": None,
            "impact_factor": None,
            "cited_by_count": None,
            "quartile": None,
            "quality": quality,
            "is_in_doaj": True,
            "has_doaj_seal": has_seal,
            "is_predatory": False,
            "predatory_source": None,
            "score_source": "doaj",
        }
        openalex_names.add(name_lower)
        doaj_added += 1

    records = list(seen.values())

    # Derive quartile (Q1–Q4) from cited_by_count percentile within each
    # source_type. Field-specific quartiles would be more accurate but
    # require per-source topic data that 4–7×s the snapshot size, so we
    # use global per-type percentiles as a defensible approximation
    # given the license constraint that ruled out SJR.
    by_type: dict[str, list[dict]] = {}
    for r in records:
        if r.get("cited_by_count") is None:
            continue  # NULL quartile for entries without citation data
        by_type.setdefault(r.get("source_type") or "", []).append(r)
    for type_records in by_type.values():
        type_records.sort(key=lambda r: r["cited_by_count"])
        n = len(type_records)
        if n == 0:
            continue
        for rank, r in enumerate(type_records):
            pct = rank / n  # 0.0 = lowest, ~1.0 = highest
            if pct >= 0.75:
                r["quartile"] = "Q1"
            elif pct >= 0.50:
                r["quartile"] = "Q2"
            elif pct >= 0.25:
                r["quartile"] = "Q3"
            else:
                r["quartile"] = "Q4"

    # Re-derive quality now that quartile is available. The first-pass
    # `quality` values above were computed without quartile and are
    # therefore suboptimal — a Q1 journal without h-index data would
    # have scored `None` (fall-through) instead of 8. The runtime filter
    # code in journal_reputation_filter.py does pass quartile, so the
    # stored column should agree with the live score.
    for r in records:
        r["quality"] = derive_quality_score(
            h_index=r.get("h_index"),
            quartile=r.get("quartile"),
            is_in_doaj=r.get("is_in_doaj") or False,
            has_doaj_seal=r.get("has_doaj_seal") or False,
            is_predatory=r.get("is_predatory") or False,
            source_type=r.get("source_type"),
        )

    logger.info(
        f"Inserting {len(records)} source records ({doaj_added} DOAJ-only)..."
    )
    for i in range(0, len(records), _BATCH_SIZE):
        session.bulk_insert_mappings(Source, records[i : i + _BATCH_SIZE])  # type: ignore[arg-type]


def _populate_institutions(session: Session, institutions: dict) -> None:
    records: list[dict] = []
    for inst_id, compact in institutions.items():
        name = (compact.get("n") or "").strip()
        if not name:
            continue
        records.append(
            {
                "openalex_id": inst_id,
                "name": name,
                "name_lower": normalize_name(name),
                "ror_id": compact.get("r"),
                "country": compact.get("c"),
                "type": compact.get("t"),
                "h_index": compact.get("h"),
                "impact_factor": compact.get("if"),
                "works_count": compact.get("w"),
                "cited_by_count": compact.get("cb"),
            }
        )
    logger.info(f"Inserting {len(records)} institution records...")
    for i in range(0, len(records), _BATCH_SIZE):
        session.bulk_insert_mappings(Institution, records[i : i + _BATCH_SIZE])  # type: ignore[arg-type]


def _populate_abbreviations(session: Session, mappings: dict) -> None:
    records: list[dict] = []
    seen: set[str] = set()
    for abbrev, full in mappings.items():
        norm = normalize_name(abbrev)
        if not norm or norm in seen:
            continue
        seen.add(norm)
        records.append({"abbrev_lower": norm, "full_name": full})
    logger.info(f"Inserting {len(records)} abbreviation records...")
    for i in range(0, len(records), _BATCH_SIZE):
        session.bulk_insert_mappings(Abbreviation, records[i : i + _BATCH_SIZE])  # type: ignore[arg-type]


# ---------------------------------------------------------------------------
# Backwards-compat shim for the old build_reference_db name
# ---------------------------------------------------------------------------


def build_reference_db(
    data_dir: Optional[Path] = None,
    output_path: Optional[Path] = None,
) -> None:
    """Deprecated alias for `build_db`."""
    build_db(data_dir=data_dir, output_path=output_path)
