"""Add first-class DAX table resources and owner inheritance.

Revision ID: 523
Revises: 522

Existing dataset resources and their historical bindings remain unchanged.
New table resources are children of one requestable dataset and carry the
exact table locator used by approval and reconciliation.
"""
from __future__ import annotations

import sqlalchemy as sa

from alembic import op
from app.core.migration_helpers import (
    id_type_for,
    normalize_fk_column_collation,
)

revision = "523"
down_revision = "522"
branch_labels = None
depends_on = None

TABLE = "data_access_resources"
EFFECT_TABLE = "data_access_compiled_effects"
RUN_TABLE = "data_access_effect_runs"
ACTIVE_UNIQUE = "uq_data_access_resources_active_locator"
PARENT_INDEX = "ix_data_access_resources_parent"
PARENT_FK = "fk_data_access_resources_parent"


def _column_exists(bind, column: str) -> bool:
    return _column_exists_for(bind, TABLE, column)


def _column_exists_for(bind, table: str, column: str) -> bool:
    return bool(bind.execute(
        sa.text(
            "SELECT 1 FROM information_schema.columns "
            "WHERE table_schema = DATABASE() AND table_name = :table "
            "AND column_name = :column LIMIT 1",
        ),
        {"table": table, "column": column},
    ).scalar())


def _index_exists(bind, index: str) -> bool:
    return _index_exists_for(bind, TABLE, index)


def _index_exists_for(bind, table: str, index: str) -> bool:
    return bool(bind.execute(
        sa.text(
            "SELECT 1 FROM information_schema.statistics "
            "WHERE table_schema = DATABASE() AND table_name = :table "
            "AND index_name = :index LIMIT 1",
        ),
        {"table": table, "index": index},
    ).scalar())


def _fk_exists(bind, name: str) -> bool:
    # Both schema predicates on purpose. ``CONSTRAINT_SCHEMA`` alone leaves
    # the TABLE side of the row unscoped, so on a server hosting a second
    # helix schema this can match a same-named constraint on another
    # database's copy of the table and skip the DDL — the silent-success
    # failure ``test_information_schema_probes_are_schema_local`` exists to
    # prevent. A FK always lives in its table's schema, so adding
    # ``TABLE_SCHEMA = DATABASE()`` strictly narrows and is a no-op for an
    # already-migrated database.
    return bool(bind.execute(
        sa.text(
            "SELECT 1 FROM information_schema.table_constraints "
            "WHERE constraint_schema = DATABASE() "
            "AND table_schema = DATABASE() AND table_name = :table "
            "AND constraint_name = :name AND constraint_type = 'FOREIGN KEY' "
            "LIMIT 1",
        ),
        {"table": TABLE, "name": name},
    ).scalar())


def _parent_fk_name(bind) -> str | None:
    # Same reasoning as ``_fk_exists`` — and here an unscoped read is worse
    # than a skipped guard: the NAME it returns is fed to ``drop_constraint``
    # below, so a cross-schema match would have us try to drop a constraint
    # that does not exist in this database.
    return bind.execute(
        sa.text(
            "SELECT constraint_name FROM information_schema.key_column_usage "
            "WHERE constraint_schema = DATABASE() "
            "AND table_schema = DATABASE() AND table_name = :table "
            "AND column_name = 'parent_resource_id' "
            "AND referenced_table_name = :table LIMIT 1",
        ),
        {"table": TABLE},
    ).scalar()


def upgrade() -> None:
    bind = op.get_bind()
    if _index_exists(bind, ACTIVE_UNIQUE):
        op.drop_index(ACTIVE_UNIQUE, table_name=TABLE)
    if _column_exists(bind, "active_locator_key"):
        op.drop_column(TABLE, "active_locator_key")

    if not _column_exists(bind, "parent_resource_id"):
        op.add_column(
            TABLE,
            sa.Column("parent_resource_id", id_type_for(TABLE), nullable=True),
        )
        normalize_fk_column_collation(
            TABLE,
            "parent_resource_id",
            TABLE,
            nullable=True,
        )
    if not _column_exists(bind, "table_id"):
        op.add_column(
            TABLE,
            sa.Column(
                "table_id",
                sa.String(1024, collation="utf8mb4_bin"),
                nullable=True,
            ),
        )
    if not _column_exists(bind, "table_type"):
        op.add_column(TABLE, sa.Column("table_type", sa.String(32), nullable=True))

    if not _fk_exists(bind, PARENT_FK):
        op.create_foreign_key(
            PARENT_FK,
            TABLE,
            TABLE,
            ["parent_resource_id"],
            ["id"],
            ondelete="CASCADE",
        )
    if not _index_exists(bind, PARENT_INDEX):
        op.create_index(
            PARENT_INDEX,
            TABLE,
            ["parent_resource_id"],
        )

    op.add_column(
        TABLE,
        sa.Column(
            "active_locator_key",
            sa.String(64),
            sa.Computed(
                "CASE WHEN status = 'active' THEN "
                "SHA2(CONCAT_WS('|', integration_id, provider, resource_type, "
                "project_id, dataset_id, COALESCE(table_id, '')), 256) "
                "ELSE NULL END",
            ),
            nullable=True,
        ),
    )
    op.create_index(
        ACTIVE_UNIQUE,
        TABLE,
        ["active_locator_key"],
        unique=True,
    )

    effect_target_index = "ix_data_access_compiled_effects_target"
    if _index_exists_for(bind, EFFECT_TABLE, effect_target_index):
        op.drop_index(effect_target_index, table_name=EFFECT_TABLE)
    if _column_exists_for(bind, EFFECT_TABLE, "target_locator_key"):
        op.drop_column(EFFECT_TABLE, "target_locator_key")
    if not _column_exists_for(bind, EFFECT_TABLE, "target_table_id"):
        op.add_column(
            EFFECT_TABLE,
            sa.Column(
                "target_table_id",
                sa.String(1024, collation="utf8mb4_bin"),
                nullable=True,
            ),
        )
    op.add_column(
        EFFECT_TABLE,
        sa.Column(
            "target_locator_key",
            sa.String(64),
            sa.Computed(
                "SHA2(CONCAT_WS('|', provider, resource_type, "
                "target_project_id, target_dataset_id, "
                "COALESCE(target_table_id, '')), 256)",
            ),
            nullable=True,
        ),
    )
    op.create_index(
        effect_target_index,
        EFFECT_TABLE,
        ["target_locator_key"],
    )
    if not _column_exists_for(bind, RUN_TABLE, "target_table_id"):
        op.add_column(
            RUN_TABLE,
            sa.Column(
                "target_table_id",
                sa.String(1024, collation="utf8mb4_bin"),
                nullable=True,
            ),
        )


def downgrade() -> None:
    bind = op.get_bind()
    table_resource_count = int(bind.execute(sa.text(
        "SELECT COUNT(*) FROM data_access_resources WHERE resource_type = 'table'",
    )).scalar() or 0)
    if table_resource_count:
        raise RuntimeError(
            "Cannot downgrade migration 523 while DAX table resources exist",
        )
    effect_target_index = "ix_data_access_compiled_effects_target"
    if _index_exists_for(bind, EFFECT_TABLE, effect_target_index):
        op.drop_index(effect_target_index, table_name=EFFECT_TABLE)
    if _column_exists_for(bind, EFFECT_TABLE, "target_locator_key"):
        op.drop_column(EFFECT_TABLE, "target_locator_key")
    if _column_exists_for(bind, EFFECT_TABLE, "target_table_id"):
        op.drop_column(EFFECT_TABLE, "target_table_id")
    op.add_column(
        EFFECT_TABLE,
        sa.Column(
            "target_locator_key",
            sa.String(64),
            sa.Computed(
                "SHA2(CONCAT_WS('|', provider, resource_type, "
                "target_project_id, target_dataset_id), 256)",
            ),
            nullable=True,
        ),
    )
    op.create_index(
        effect_target_index,
        EFFECT_TABLE,
        ["target_locator_key"],
    )
    if _column_exists_for(bind, RUN_TABLE, "target_table_id"):
        op.drop_column(RUN_TABLE, "target_table_id")
    if _index_exists(bind, ACTIVE_UNIQUE):
        op.drop_index(ACTIVE_UNIQUE, table_name=TABLE)
    if _column_exists(bind, "active_locator_key"):
        op.drop_column(TABLE, "active_locator_key")
    parent_fk_name = _parent_fk_name(bind)
    if parent_fk_name:
        op.drop_constraint(str(parent_fk_name), TABLE, type_="foreignkey")
    if _index_exists(bind, PARENT_INDEX):
        op.drop_index(PARENT_INDEX, table_name=TABLE)
    for column in ("table_type", "table_id", "parent_resource_id"):
        if _column_exists(bind, column):
            op.drop_column(TABLE, column)
    op.add_column(
        TABLE,
        sa.Column(
            "active_locator_key",
            sa.String(64),
            sa.Computed(
                "CASE WHEN status = 'active' THEN "
                "SHA2(CONCAT_WS('|', integration_id, provider, resource_type, "
                "project_id, dataset_id), 256) ELSE NULL END",
            ),
            nullable=True,
        ),
    )
    op.create_index(
        ACTIVE_UNIQUE,
        TABLE,
        ["active_locator_key"],
        unique=True,
    )