"""Journal quality system — unified schema migration.

Single migration for the journal-quality feature that combines what was
originally split across 0006–0010 during development:

- ``papers`` and ``paper_appearances`` tables for deduplicated citation
  metadata (originally migration 0006).
- Three columns on ``journals`` — ``name_lower``, ``score_source``,
  ``quality_model`` — plus the two indexes that serve the case-insensitive
  cache lookup hot path (originally migration 0007). ``name_lower`` is
  backfilled from existing ``name`` values using Python ``str.lower()`` so
  diacritics are handled identically to the runtime insert path.
- ``ix_research_resources_research_id`` index on the FK used by every
  research-detail join (originally migration 0009).

The feature never shipped, so consolidating the five revisions into one is
safe: no production database is stamped at any of the intermediate ids.
Dev databases on the feature branch stamped at 0006–0010 will need to be
reset (delete the file and let the app re-initialize) after pulling this
change.

Revision ID: 0006
Revises: 0005
Create Date: 2026-04-18
"""

import unicodedata

from alembic import op
import sqlalchemy as sa
from sqlalchemy import inspect
from sqlalchemy_utc import UtcDateTime

# revision identifiers, used by Alembic.
revision = "0006"
down_revision = "0005"
branch_labels = None
depends_on = None


# --------------------------------------------------------------------------- #
# Introspection helpers — keep the migration idempotent on reruns.             #
# --------------------------------------------------------------------------- #


def _table_exists(table_name: str) -> bool:
    bind = op.get_bind()
    return inspect(bind).has_table(table_name)


def _column_exists(table_name: str, column_name: str) -> bool:
    bind = op.get_bind()
    inspector = inspect(bind)
    if not inspector.has_table(table_name):
        return False
    return column_name in {
        col["name"] for col in inspector.get_columns(table_name)
    }


def _index_exists(table_name: str, index_name: str) -> bool:
    bind = op.get_bind()
    inspector = inspect(bind)
    if not inspector.has_table(table_name):
        return False
    return index_name in {
        ix["name"] for ix in inspector.get_indexes(table_name)
    }


def _unique_constraint_exists(table_name: str, constraint_name: str) -> bool:
    bind = op.get_bind()
    inspector = inspect(bind)
    if not inspector.has_table(table_name):
        return False
    return constraint_name in {
        uc.get("name") for uc in inspector.get_unique_constraints(table_name)
    }


# Columns added to ``journals`` by this migration. Existing ``quality`` and
# ``quality_analysis_time`` columns from the 0001 baseline are untouched.
_JOURNAL_COLUMNS = [
    ("name_lower", sa.String(255), {"nullable": True}),
    ("score_source", sa.String(50), {"nullable": True}),
    ("quality_model", sa.String(255), {"nullable": True}),
]

_JOURNAL_INDEXES = [
    # NB: no separate index on ``name_lower``. The
    # ``uq_journals_name_lower`` UNIQUE constraint created below
    # provides a backing index, and a second non-unique index on the
    # same column would be pure duplication — two B-trees maintained
    # per INSERT/UPDATE for no query benefit.
    ("ix_journals_quality_model", ["quality_model"]),
]


def upgrade() -> None:
    # --- papers table (one row per unique academic paper) ---
    # Minimal schema: only columns used for dedup lookups and dashboard
    # joins are real. All bibliographic fields (authors, year, volume,
    # container_title, csl_json, ...) go into the ``metadata`` JSON blob
    # to avoid NULL bloat and schema rigidity.
    if not _table_exists("papers"):
        op.create_table(
            "papers",
            sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True),
            # Academic identifiers (waterfall dedup keys). UNIQUE constraints
            # prevent concurrent writers from creating duplicate rows. SQLite
            # allows multiple NULLs in a UNIQUE column (standard SQL), so
            # papers without identifiers are still permitted.
            sa.Column("doi", sa.String(255), nullable=True, unique=True),
            sa.Column("arxiv_id", sa.String(100), nullable=True, unique=True),
            sa.Column("pmid", sa.String(50), nullable=True, unique=True),
            sa.Column(
                "journal_id",
                sa.Integer(),
                sa.ForeignKey("journals.id", ondelete="SET NULL"),
                nullable=True,
            ),
            # Cleaned journal name that matched the filter's scoring
            # tiers (post-regex-clean + abbreviation-expand + optional
            # LLM-relabel). Indexed so the dashboard can GROUP BY it
            # and batch-enrich from the shared read-only reference DB.
            sa.Column("container_title", sa.String(500), nullable=True),
            # NOTE: no ``journal_quality`` column by design. Quality is
            # resolved live in the dashboard — Tier 4 via the user's
            # ``journals`` table (keyed by NFKC-normalized
            # ``container_title``) and Tier 1-3 via the bundled
            # read-only reference DB. A frozen per-Paper copy would go
            # stale when a journal is re-scored (new LLM model, bug
            # fix, manual override). Don't add it back. See Paper
            # model docstring in database/models/citation.py.
            # Publication year — promoted out of the metadata JSON blob
            # to a first-class integer column so the dashboard can
            # filter/group/sort by year without paying for json_extract
            # on every row. Always written alongside the JSON copy in
            # paper_metadata so existing readers keep working; the
            # column is a denormalized index surface.
            sa.Column("year", sa.Integer(), nullable=True),
            sa.Column("metadata", sa.JSON(), nullable=True),
            # Timestamps use Python-side defaults (matches the Paper
            # model's ``default=utcnow()`` + ``onupdate=utcnow()`` at
            # citation.py:102-105). A SQL-level server_default would
            # drift from the fresh-install path — 0001 creates this
            # table via create_all() from the model, which omits
            # server-side defaults, so migration-replay + create_all
            # paths would produce inconsistent schemas.
            sa.Column("created_at", UtcDateTime(), nullable=False),
            sa.Column("updated_at", UtcDateTime(), nullable=False),
        )
    # Index creation runs unconditionally (guarded by _index_exists) so a
    # rerun where ``papers`` already exists but is missing one of the
    # named indexes — e.g. a partially-applied prior migration or a DB
    # that was manually patched — still converges to the full schema.
    # doi / arxiv_id / pmid each carry UNIQUE, which already creates a
    # backing unique index — no separate non-unique index needed.
    if _table_exists("papers"):
        if not _index_exists("papers", "idx_papers_journal"):
            op.create_index("idx_papers_journal", "papers", ["journal_id"])
        # container_title is the dashboard's GROUP BY key — indexed.
        if not _index_exists("papers", "idx_papers_container_title"):
            op.create_index(
                "idx_papers_container_title", "papers", ["container_title"]
            )
        if not _index_exists("papers", "idx_papers_year"):
            op.create_index("idx_papers_year", "papers", ["year"])

    # --- paper_appearances table (links papers to research resources) ---
    if not _table_exists("paper_appearances"):
        op.create_table(
            "paper_appearances",
            sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True),
            sa.Column(
                "paper_id",
                sa.Integer(),
                sa.ForeignKey("papers.id", ondelete="CASCADE"),
                nullable=False,
            ),
            sa.Column(
                "resource_id",
                sa.Integer(),
                sa.ForeignKey("research_resources.id", ondelete="CASCADE"),
                nullable=False,
                unique=True,
            ),
            sa.Column("source_engine", sa.String(50), nullable=True),
            # Python-side default — matches citation.py:157 and the
            # papers table above. See the papers block for the full
            # rationale.
            sa.Column("created_at", UtcDateTime(), nullable=False),
        )
    # Explicit index creation — alembic upgrade does NOT run
    # Base.metadata.create_all, so ``index=True`` on the model column
    # is not enough for the migration-replay path. The model declares
    # ``index=True`` on PaperAppearance.paper_id (citation.py:159),
    # which produces the same name ``ix_paper_appearances_paper_id`` on
    # the create_all() path; name match is what lets Alembic autogenerate
    # reconcile the two without drift.
    # resource_id carries UNIQUE above, which already creates a backing
    # unique index — no separate index needed.
    if _table_exists("paper_appearances") and not _index_exists(
        "paper_appearances", "ix_paper_appearances_paper_id"
    ):
        op.create_index(
            "ix_paper_appearances_paper_id",
            "paper_appearances",
            ["paper_id"],
        )

    # --- journals extensions (name_lower + score_source + quality_model) ---
    # All three columns, plus the two indexes, go into a single
    # ``batch_alter_table`` so SQLite only rebuilds the table once.
    if _table_exists("journals"):
        columns_to_add = [
            (name, col_type, kwargs)
            for name, col_type, kwargs in _JOURNAL_COLUMNS
            if not _column_exists("journals", name)
        ]
        indexes_to_create = [
            (idx_name, idx_columns)
            for idx_name, idx_columns in _JOURNAL_INDEXES
            if not _index_exists("journals", idx_name)
        ]

        bind = op.get_bind()

        # Step 1 (BEFORE column add): dedupe rows that would collide
        # on ``name_lower`` once backfilled. Two pre-existing rows
        # whose ``name`` values differ only in case or Unicode
        # compatibility form (e.g. "Nature Medicine" vs "NATURE
        # MEDICINE", or "Physics Letters™" vs "Physics LettersTM")
        # would share the same ``name_lower`` and trip the new UNIQUE
        # constraint during ``batch_alter_table``.
        #
        # Normalization MUST match scoring.normalize_name (NFKC + lower
        # + strip). We inline the expression rather than importing
        # normalize_name because migrations should not depend on
        # application-layer modules that may be reorganized; if
        # scoring.py changes normalize semantics, this inline must be
        # updated in lockstep — there is no automatic coupling.
        #
        # Tiebreaker: keep the HIGHEST quality row (best LLM verdict
        # wins), with ties broken by lowest id. NOT first-writer-wins
        # — that would discard a quality=9 row in favour of an older
        # quality=5 row. NULL quality sorts last so any scored row
        # beats an unscored one.
        #
        # Doing this BEFORE the column add works around SQLite's
        # behaviour of enforcing new UNIQUE constraints during the
        # table-copy step of batch_alter_table (copy fails on
        # pre-existing duplicates; dedupe-post-copy comes too late).
        # Safe: the Journal cache is reproducible — the next LLM pass
        # recreates any pruned rows.
        #
        # Concurrency note: BEGIN EXCLUSIVE was considered here and
        # rejected — it raises "cannot start a transaction within a
        # transaction" inside the outer engine.begin() that
        # alembic_runner.py:268 opens. In practice, SQLite WAL +
        # _connections_lock (encrypted_db.py) + the current==head
        # short-circuit in alembic_runner serialize access sufficiently.
        journals_pre = sa.Table("journals", sa.MetaData(), autoload_with=bind)
        all_rows = bind.execute(
            sa.select(
                journals_pre.c.id,
                journals_pre.c.name,
                journals_pre.c.quality,
            )
        ).fetchall()
        from collections import defaultdict as _dd

        groups: dict = _dd(list)
        for row in all_rows:
            if row.name is not None:
                name_lower_canonical = (
                    unicodedata.normalize("NFKC", row.name).lower().strip()
                )
                groups[name_lower_canonical].append((row.id, row.quality))
        ids_to_delete = [
            lose_id
            for rows in groups.values()
            if len(rows) > 1
            for lose_id, _ in sorted(rows, key=lambda r: (-(r[1] or 0), r[0]))[
                1:
            ]
        ]
        if ids_to_delete:
            bind.execute(
                journals_pre.delete().where(
                    journals_pre.c.id.in_(ids_to_delete)
                )
            )

        # Step 2: add the new columns + indexes. The model's UNIQUE on
        # ``name_lower`` is now safe to apply because Step 1 removed
        # all potential collisions.
        if columns_to_add or indexes_to_create:
            with op.batch_alter_table("journals") as batch_op:
                for col_name, col_type, kwargs in columns_to_add:
                    batch_op.add_column(sa.Column(col_name, col_type, **kwargs))
                for idx_name, idx_columns in indexes_to_create:
                    batch_op.create_index(idx_name, idx_columns)

        # Step 3: backfill ``name_lower``. Re-reflect the table so the
        # newly-added column is visible. Normalization MUST match the
        # Step 1 dedupe expression above (and scoring.normalize_name)
        # — NFKC + lower + strip in Python, not SQL ``LOWER()`` which
        # would leave Unicode compatibility characters intact.
        journals_tbl = sa.Table("journals", sa.MetaData(), autoload_with=bind)
        rows = bind.execute(
            sa.select(journals_tbl.c.id, journals_tbl.c.name).where(
                journals_tbl.c.name_lower.is_(None),
                journals_tbl.c.name.is_not(None),
            )
        ).fetchall()
        for i in range(0, len(rows), 500):
            batch = [
                {
                    "b_id": r.id,
                    "b_name_lower": unicodedata.normalize("NFKC", r.name or "")
                    .lower()
                    .strip(),
                }
                for r in rows[i : i + 500]
            ]
            if not batch:
                continue
            bind.execute(
                journals_tbl.update()
                .where(journals_tbl.c.id == sa.bindparam("b_id"))
                .values(name_lower=sa.bindparam("b_name_lower")),
                batch,
            )

        # Step 4: add the UNIQUE constraint. Idempotent via
        # ``_unique_constraint_exists`` — re-runs are no-ops.
        if not _unique_constraint_exists("journals", "uq_journals_name_lower"):
            with op.batch_alter_table("journals") as batch_op:
                batch_op.create_unique_constraint(
                    "uq_journals_name_lower", ["name_lower"]
                )

    # --- research_resources: index the FK used by every join ---
    if (
        _table_exists("research_resources")
        and _column_exists("research_resources", "research_id")
        and not _index_exists(
            "research_resources", "ix_research_resources_research_id"
        )
    ):
        with op.batch_alter_table("research_resources") as batch_op:
            batch_op.create_index(
                "ix_research_resources_research_id",
                ["research_id"],
            )


def downgrade() -> None:
    """Reverse the 0006 schema changes.

    DATA LOSS WARNING: this drops the ``papers`` and ``paper_appearances``
    tables and the ``name_lower`` / ``score_source`` / ``quality_model``
    columns on ``journals``. Every row in those tables and every value
    in those columns is unrecoverable after downgrade — only the
    ``name``, ``quality``, ``id``, and ``quality_analysis_time``
    columns on ``journals`` survive. Callers needing to preserve the
    data must back it up before running this downgrade.
    """
    # Inverse order: research_resources index → journals columns/indexes →
    # paper_appearances → papers.
    if _table_exists("research_resources") and _index_exists(
        "research_resources", "ix_research_resources_research_id"
    ):
        with op.batch_alter_table("research_resources") as batch_op:
            batch_op.drop_index("ix_research_resources_research_id")

    if _table_exists("journals"):
        # KNOWN-DEFERRED: uq_journals_name_lower is not explicitly
        # dropped here because SQLite batch_alter_table rebuilds the
        # whole table from scratch when a column is dropped, discarding
        # all constraints naturally. On Postgres this would require an
        # explicit drop_constraint call before drop_column. Tracked as
        # a portability follow-up.
        with op.batch_alter_table("journals") as batch_op:
            for idx_name, _ in _JOURNAL_INDEXES:
                if _index_exists("journals", idx_name):
                    batch_op.drop_index(idx_name)
            for col_name, _, _ in _JOURNAL_COLUMNS:
                if _column_exists("journals", col_name):
                    batch_op.drop_column(col_name)

    if _table_exists("paper_appearances"):
        op.drop_table("paper_appearances")
    if _table_exists("papers"):
        op.drop_table("papers")
