"""Data-access BigQuery dataset resource projection.

Revision ID: 328
Revises: 327
Create Date: 2026-06-10

DAX-006 creates the first durable data-access table:
``data_access_resources``. It projects BigQuery dataset identity from the
catalog or ops-bootstrap input into Helix-owned runtime state. Request and
revoke authorization may read resource-owner authority only from this row,
never from raw catalog/bootstrap input.

The table deliberately stores provider/resource vocabulary as canonical
strings from DAX-002 and keeps owner provenance explicit:

* ``owner_person_id`` is nullable. It grants resource-owner authority only
  when ``owner_resolution_status='resolved'``.
* ``owner_resolution_status`` can represent ``unresolved_owner`` and
  ``fallback_owner`` rows without granting authority.
* ``owner_source`` + ``source_version`` + ``source_etag`` carry
  catalog/bootstrap provenance for staleness checks and cutover.

MySQL/TiDB do not support PostgreSQL-style partial unique indexes, so active
locator uniqueness uses a generated hash key. Only ``status='active'`` rows
produce ``active_locator_key``; suspended/deleted historical rows produce
NULL and can coexist for audit/cutover history.
"""
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 = "331"
down_revision = "330"
branch_labels = None
depends_on = None

TABLE = "data_access_resources"
ACTIVE_UNIQUE = "uq_data_access_resources_active_locator"


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

    resource_id = id_type_for("identity_persons")
    integration_id = id_type_for("integrations")
    person_id = id_type_for("identity_persons")
    kw = table_kw_for("identity_persons")

    op.create_table(
        TABLE,
        sa.Column("id", resource_id, primary_key=True),
        sa.Column("integration_id", integration_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(
            "project_id",
            sa.String(64, collation="utf8mb4_bin"),
            nullable=False,
        ),
        sa.Column(
            "dataset_id",
            sa.String(1024, collation="utf8mb4_bin"),
            nullable=False,
        ),
        sa.Column("display_name", sa.String(255), nullable=True),
        sa.Column(
            "requestable", sa.Boolean(), nullable=False,
            server_default="1",
        ),
        sa.Column(
            "status", sa.String(24), nullable=False,
            server_default="active",
        ),
        sa.Column("owner_person_id", person_id, nullable=True),
        sa.Column(
            "owner_resolution_status", sa.String(32), nullable=False,
            server_default="resolved",
        ),
        sa.Column("owner_source", sa.String(32), nullable=False),
        sa.Column("source_version", sa.String(128), nullable=False),
        sa.Column("source_etag", sa.String(256), nullable=False),
        sa.Column("source_payload_digest", sa.String(128), nullable=True),
        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,
        ),
        sa.Column("notes", 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,
    )
    normalize_fk_column_collation(
        TABLE, "integration_id", "integrations", nullable=False,
    )
    normalize_fk_column_collation(
        TABLE, "owner_person_id", "identity_persons", nullable=True,
    )
    op.create_foreign_key(
        "fk_data_access_resources_integration",
        TABLE,
        "integrations",
        ["integration_id"],
        ["id"],
        ondelete="RESTRICT",
    )
    op.create_foreign_key(
        "fk_data_access_resources_owner",
        TABLE,
        "identity_persons",
        ["owner_person_id"],
        ["id"],
        ondelete="SET NULL",
    )
    op.create_index(
        ACTIVE_UNIQUE,
        TABLE,
        ["active_locator_key"],
        unique=True,
    )
    op.create_index(
        "ix_data_access_resources_integration", TABLE, ["integration_id"],
    )
    op.create_index(
        "ix_data_access_resources_owner", TABLE, ["owner_person_id"],
    )
    op.create_index("ix_data_access_resources_status", TABLE, ["status"])
    op.create_index(
        "ix_data_access_resources_requestable", TABLE, ["requestable"],
    )
    op.create_index(
        "ix_data_access_resources_source",
        TABLE,
        ["owner_source", "source_version"],
    )


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