"""Identity consumers — field-level authorization grant.

Adds ``identity_consumers.granted_fields`` (JSON list of person-field
keys). The three coarse scopes survive as bundles; this column is the
least-privilege projection the ``/persons`` serializer applies.

Backfill is point-in-time and self-contained (the bundle→field mapping
is inlined, not imported from app code, so a later catalog change can't
rewrite history): every pre-existing consumer expands its current
``scopes`` to the full field set of those bundles, so behaviour is
unchanged for consumers that only ever ticked bundles.

Two-step add (nullable → backfill → NOT NULL) keeps the column NOT NULL
to match the ORM model while staying portable across MySQL 8 and TiDB
(no reliance on a JSON expression DEFAULT for existing rows).
"""
from __future__ import annotations

import json

import sqlalchemy as sa

from alembic import op

revision = "285"
down_revision = "284"
branch_labels = None
depends_on = None

CONSUMERS = "identity_consumers"

# Point-in-time bundle → person-field mapping (see field_catalog.py).
_PERSONS_FIELDS = [
    "id", "updated_at", "employee_no", "work_email", "display_name",
    "first_name", "last_name", "avatar_url", "employment_status",
    "job_title", "is_china_employee", "department", "manager", "hrbp",
    "work_location", "company", "cost_center", "hire_date",
    "termination_date", "preferred_language", "email_aliases",
    "source_updated_at",
]
_PII_FIELDS = ["mobile_phone", "personal_email"]


def _default_fields(scopes: list[str]) -> list[str]:
    out: list[str] = []
    if "directory.persons:read" in scopes:
        out += _PERSONS_FIELDS
    if "directory.pii:read" in scopes:
        out += _PII_FIELDS
    return out


def _column_exists(bind, table: str, column: str) -> bool:
    return bind.execute(
        sa.text(
            "SELECT 1 FROM information_schema.COLUMNS "
            "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :t "
            "AND COLUMN_NAME = :c"
        ),
        {"t": table, "c": column},
    ).scalar() is not None


def upgrade() -> None:
    bind = op.get_bind()
    if not _column_exists(bind, CONSUMERS, "granted_fields"):
        op.add_column(
            CONSUMERS, sa.Column("granted_fields", sa.JSON(), nullable=True),
        )

    rows = bind.execute(
        sa.text(f"SELECT id, scopes FROM {CONSUMERS}")  # noqa: S608 — constant
    ).fetchall()
    for row in rows:
        raw = row[1]
        scopes = raw if isinstance(raw, list) else json.loads(raw or "[]")
        bind.execute(
            sa.text(
                f"UPDATE {CONSUMERS} SET granted_fields = :gf WHERE id = :id"  # noqa: S608
            ),
            {"gf": json.dumps(_default_fields(scopes)), "id": row[0]},
        )

    op.alter_column(
        CONSUMERS, "granted_fields",
        existing_type=sa.JSON(), nullable=False,
    )


def downgrade() -> None:
    if _column_exists(op.get_bind(), CONSUMERS, "granted_fields"):
        op.drop_column(CONSUMERS, "granted_fields")