"""Data-access compiled provider effects.

Revision ID: 332
Revises: 331
Create Date: 2026-06-10

DAX-010 adds ``data_access_compiled_effects``. These rows are the
provider-specific output compiled from semantic binding state; they do
not run a reconciler or mutate BigQuery.

The semantic principal remains ``email_groups.id``. ``provider_member``
persists the compiled provider member snapshot such as
``group:analytics@example.com`` so group-primary-email rename handling
can compare and intentionally create old-email absent / new-email
present effects. Aliases are not stored.

Each effect carries provenance needed by stale/cutover/recompile checks:
entitlement version, resource source version/etag, compiler version,
target locator, managed reason, status, and the first-apply adoption
marker ``already_present_at_first_apply``.
"""
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,
    table_kw_for,
)

revision = "335"
down_revision = "334"
branch_labels = None
depends_on = None

TABLE = "data_access_compiled_effects"


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


def upgrade() -> None:
    bind = op.get_bind()
    if _table_exists(bind, TABLE):
        return

    row_id = id_type_for("identity_persons")
    binding_id = id_type_for("data_access_bindings")
    resource_id = id_type_for("data_access_resources")
    group_id = id_type_for("email_groups")
    entitlement_id = id_type_for("data_access_entitlements")
    kw = table_kw_for("identity_persons")

    op.create_table(
        TABLE,
        sa.Column("id", row_id, primary_key=True),
        sa.Column("binding_id", binding_id, nullable=False),
        sa.Column("resource_id", resource_id, nullable=False),
        sa.Column("principal_email_group_id", group_id, nullable=False),
        sa.Column("entitlement_id", entitlement_id, nullable=False),
        sa.Column(
            "provider", sa.String(32), nullable=False,
            server_default="bigquery",
        ),
        sa.Column(
            "resource_type", sa.String(32), nullable=False,
            server_default="dataset",
        ),
        sa.Column(
            "target_project_id",
            sa.String(64, collation="utf8mb4_bin"),
            nullable=False,
        ),
        sa.Column(
            "target_dataset_id",
            sa.String(1024, collation="utf8mb4_bin"),
            nullable=False,
        ),
        sa.Column("provider_member", sa.String(320), nullable=False),
        sa.Column(
            "provider_role", sa.String(128), nullable=False,
            server_default="roles/bigquery.dataViewer",
        ),
        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,
        ),
        sa.Column(
            "desired_state", sa.String(16), nullable=False,
            server_default="present",
        ),
        sa.Column(
            "entitlement_version_snapshot",
            sa.String(16),
            nullable=False,
        ),
        sa.Column(
            "resource_source_version_snapshot",
            sa.String(128),
            nullable=False,
        ),
        sa.Column(
            "resource_source_etag_snapshot",
            sa.String(256),
            nullable=False,
        ),
        sa.Column("compiler_version", sa.String(64), nullable=False),
        sa.Column("managed_reason", sa.String(64), nullable=False),
        sa.Column(
            "already_present_at_first_apply",
            sa.Boolean(),
            nullable=False,
            server_default="0",
        ),
        sa.Column(
            "status", sa.String(32), nullable=False,
            server_default="pending_reconcile",
        ),
        sa.Column("last_provider_etag", sa.String(256), nullable=True),
        sa.Column("last_provider_outcome", sa.String(64), nullable=True),
        sa.Column("last_error", sa.Text(), nullable=True),
        sa.Column(
            "created_at", sa.DateTime(timezone=True), nullable=False,
            server_default=sa.func.now(),
        ),
        sa.Column(
            "updated_at", sa.DateTime(timezone=True), nullable=False,
            server_default=sa.func.now(),
        ),
        **kw,
    )
    for column, parent in (
        ("binding_id", "data_access_bindings"),
        ("resource_id", "data_access_resources"),
        ("principal_email_group_id", "email_groups"),
        ("entitlement_id", "data_access_entitlements"),
    ):
        normalize_fk_column_collation(TABLE, column, parent, nullable=False)
    op.create_foreign_key(
        "fk_data_access_compiled_effects_binding",
        TABLE,
        "data_access_bindings",
        ["binding_id"],
        ["id"],
    )
    op.create_foreign_key(
        "fk_data_access_compiled_effects_resource",
        TABLE,
        "data_access_resources",
        ["resource_id"],
        ["id"],
    )
    op.create_foreign_key(
        "fk_data_access_compiled_effects_principal_group",
        TABLE,
        "email_groups",
        ["principal_email_group_id"],
        ["id"],
    )
    op.create_foreign_key(
        "fk_data_access_compiled_effects_entitlement",
        TABLE,
        "data_access_entitlements",
        ["entitlement_id"],
        ["id"],
    )
    op.create_index(
        "ix_data_access_compiled_effects_binding", TABLE, ["binding_id"],
    )
    op.create_index(
        "ix_data_access_compiled_effects_resource", TABLE, ["resource_id"],
    )
    op.create_index(
        "ix_data_access_compiled_effects_principal",
        TABLE,
        ["principal_email_group_id"],
    )
    op.create_index(
        "ix_data_access_compiled_effects_status", TABLE, ["status"],
    )
    op.create_index(
        "ix_data_access_compiled_effects_target",
        TABLE,
        ["target_locator_key"],
    )


def downgrade() -> None:
    bind = op.get_bind()
    if not _table_exists(bind, TABLE):
        return
    op.drop_table(TABLE)