from __future__ import annotations

import uuid
from datetime import UTC, datetime
from pathlib import Path
from types import SimpleNamespace
from typing import Any

from sqlalchemy import select

from app.domains.data_access import constants as c
from app.domains.data_access.models import (
    BINDING_STATUS_ACTIVE,
    BINDING_STATUS_REVOKED_PENDING,
    EFFECT_DESIRED_PRESENT,
    EFFECT_MANAGED_REASON_GRANT_REQUEST,
    EFFECT_STATUS_APPLIED,
    EFFECT_STATUS_PENDING_RECONCILE,
    ENTITLEMENT_STATUS_ACTIVE,
    ENTITLEMENT_VERSION_V1,
    OWNER_RESOLUTION_RESOLVED,
    OWNER_SOURCE_CATALOG,
    RECONCILE_STATUS_APPLIED,
    RECONCILE_STATUS_PENDING,
    REQUEST_ITEM_STATUS_APPROVED,
    REQUEST_STATUS_APPROVED,
    RESOURCE_STATUS_ACTIVE,
    DataAccessBinding,
    DataAccessCompiledEffect,
    DataAccessEffectRun,
    DataAccessEntitlement,
    DataAccessRequest,
    DataAccessRequestItem,
    DataAccessResource,
)
from app.domains.email_group.models import EmailGroup, EmailGroupOwner
from app.integrations.driver_base import (
    DataAccessDatasetIamBinding,
    DataAccessDatasetIamPolicy,
    DataAccessDatasetIamPolicyPatch,
    DataAccessDatasetIamUpdateResult,
    DataAccessDriver,
)
from app.integrations.models import Integration
from app.scheduled_jobs.data_access_handlers import (
    DATA_ACCESS_BIGQUERY_APPLY_HANDLER,
    _collect_result_failures,
    run_data_access_bigquery_apply_reconcile,
)
from app.scheduled_jobs.registry import get_handler


class _FakeApplyDriver(DataAccessDriver):
    provider = c.PROVIDER_BIGQUERY

    def __init__(self) -> None:
        super().__init__(integration_id="fake-data-access-integration", config={})
        self.get_calls: list[tuple[str, str]] = []
        self.set_calls: list[
            tuple[str, str, str | None, DataAccessDatasetIamPolicyPatch]
        ] = []

    async def get_dataset_iam(
        self,
        project_id: str,
        dataset_id: str,
    ) -> DataAccessDatasetIamPolicy:
        self.get_calls.append((project_id, dataset_id))
        return DataAccessDatasetIamPolicy(
            project_id=project_id,
            dataset_id=dataset_id,
            etag="etag-before",
            bindings=(),
            raw_policy={"etag": "etag-before"},
        )

    async def set_dataset_iam(
        self,
        project_id: str,
        dataset_id: str,
        expected_etag: str | None,
        policy_patch: DataAccessDatasetIamPolicyPatch,
    ) -> DataAccessDatasetIamUpdateResult:
        self.set_calls.append((project_id, dataset_id, expected_etag, policy_patch))
        return DataAccessDatasetIamUpdateResult(
            ok=True,
            changed=True,
            etag="etag-after",
        )


async def test_data_access_apply_handler_drains_pending_present_effect(
    db_session,
    employee_user,
    monkeypatch,
) -> None:
    requester, _ = employee_user
    integration = Integration(
        kind=c.DOMAIN_DATA_ACCESS,
        provider=c.PROVIDER_BIGQUERY,
        name="DAX BigQuery",
        slug=f"dax-bq-scheduled-{uuid.uuid4().hex[:8]}",
        enabled=True,
        is_primary=False,
        config_public={"apply_mode_enabled": True},
    )
    entitlement = DataAccessEntitlement(
        domain=c.DOMAIN_DATA_ACCESS,
        provider=c.PROVIDER_BIGQUERY,
        resource_type=c.RESOURCE_TYPE_DATASET,
        principal_type=c.PRINCIPAL_TYPE_GOOGLE_GROUP,
        entitlement=c.ENTITLEMENT_READ,
        version=ENTITLEMENT_VERSION_V1,
        provider_member_prefix=c.BIGQUERY_GROUP_MEMBER_PREFIX,
        provider_role=c.BIGQUERY_DATA_VIEWER_ROLE,
        requestable=True,
        status=ENTITLEMENT_STATUS_ACTIVE,
    )
    group = EmailGroup(
        email="scheduled-dax-readers@example.com",
        name="scheduled-dax-readers",
        status="active",
        google_group_id="scheduled-dax-google-group",
    )
    db_session.add_all([integration, entitlement, group])
    await db_session.flush()
    db_session.add(
        EmailGroupOwner(
            email_group_id=group.id,
            person_id=requester.id,
        ),
    )
    resource = DataAccessResource(
        integration_id=integration.id,
        provider=c.PROVIDER_BIGQUERY,
        resource_type=c.RESOURCE_TYPE_DATASET,
        project_id="gcp-digital-bi",
        dataset_id="dwh_core",
        display_name="Warehouse Core",
        requestable=True,
        status=RESOURCE_STATUS_ACTIVE,
        owner_person_id=str(requester.id),
        owner_resolution_status=OWNER_RESOLUTION_RESOLVED,
        owner_source=OWNER_SOURCE_CATALOG,
        source_version="repo-test-v1",
        source_etag="repo-test-etag-v1",
    )
    db_session.add(resource)
    await db_session.flush()
    request = DataAccessRequest(
        requester_person_id=requester.id,
        principal_email_group_id=group.id,
        principal_google_group_id_snapshot=group.google_group_id,
        principal_email_snapshot=group.email,
        entitlement_id=entitlement.id,
        entitlement_snapshot=c.ENTITLEMENT_READ,
        entitlement_version_snapshot=ENTITLEMENT_VERSION_V1,
        approval_flow_key=c.APPROVAL_FLOW_DATA_ACCESS_GROUP_GRANT,
        feishu_instance_code=f"dax-scheduled-{uuid.uuid4().hex}",
        idempotency_key=f"dax-scheduled-{uuid.uuid4().hex}",
        business_reason="scheduled handler regression",
        status=REQUEST_STATUS_APPROVED,
        reconcile_status=RECONCILE_STATUS_PENDING,
        decided_at=datetime.now(UTC),
        trace_id="trace-dax-scheduled",
    )
    db_session.add(request)
    await db_session.flush()
    item = DataAccessRequestItem(
        request_id=request.id,
        resource_id=resource.id,
        line_no=1,
        resource_project_id_snapshot=resource.project_id,
        resource_dataset_id_snapshot=resource.dataset_id,
        resource_source_version_snapshot=resource.source_version,
        resource_source_etag_snapshot=resource.source_etag,
        status=REQUEST_ITEM_STATUS_APPROVED,
    )
    db_session.add(item)
    await db_session.flush()
    binding = DataAccessBinding(
        resource_id=resource.id,
        principal_email_group_id=group.id,
        principal_google_group_id_snapshot=group.google_group_id,
        principal_email_snapshot=group.email,
        entitlement_id=entitlement.id,
        entitlement_snapshot=c.ENTITLEMENT_READ,
        entitlement_version_snapshot=ENTITLEMENT_VERSION_V1,
        source_request_id=request.id,
        source_request_item_id=item.id,
        status=BINDING_STATUS_ACTIVE,
        reconcile_status=RECONCILE_STATUS_PENDING,
    )
    db_session.add(binding)
    await db_session.flush()
    effect = DataAccessCompiledEffect(
        binding_id=binding.id,
        resource_id=resource.id,
        principal_email_group_id=group.id,
        entitlement_id=entitlement.id,
        provider=c.PROVIDER_BIGQUERY,
        resource_type=c.RESOURCE_TYPE_DATASET,
        target_project_id=resource.project_id,
        target_dataset_id=resource.dataset_id,
        provider_member=f"{c.BIGQUERY_GROUP_MEMBER_PREFIX}{group.email}",
        provider_role=c.BIGQUERY_DATA_VIEWER_ROLE,
        desired_state=EFFECT_DESIRED_PRESENT,
        entitlement_version_snapshot=ENTITLEMENT_VERSION_V1,
        resource_source_version_snapshot=resource.source_version,
        resource_source_etag_snapshot=resource.source_etag,
        compiler_version="dax-compiler-v1",
        managed_reason=EFFECT_MANAGED_REASON_GRANT_REQUEST,
        status=EFFECT_STATUS_PENDING_RECONCILE,
    )
    db_session.add(effect)
    await db_session.flush()

    fake_driver = _FakeApplyDriver()

    async def _fake_build_driver_by_id(
        integration_id: str,
        *,
        session: Any,
    ) -> DataAccessDriver:
        assert integration_id == str(integration.id)
        assert session is db_session
        return fake_driver

    import app.domains.data_access.reconciler as reconciler

    monkeypatch.setattr(reconciler, "build_driver_by_id", _fake_build_driver_by_id)

    summary = await run_data_access_bigquery_apply_reconcile(db_session, {})

    assert summary["integration_count"] == 1
    assert summary["apply_enabled_integration_count"] == 1
    assert summary["present"]["selected_present_effect_count"] == 1
    assert summary["present"]["applied_count"] == 1
    assert summary["present"]["provider_write_count"] == 1
    assert summary["absent"]["selected_absent_effect_count"] == 0
    assert fake_driver.get_calls == [("gcp-digital-bi", "dwh_core")]
    assert fake_driver.set_calls == [(
        "gcp-digital-bi",
        "dwh_core",
        "etag-before",
        DataAccessDatasetIamPolicyPatch(
            add_bindings=(
                DataAccessDatasetIamBinding(
                    role=c.BIGQUERY_DATA_VIEWER_ROLE,
                    members=("group:scheduled-dax-readers@example.com",),
                ),
            ),
            remove_bindings=(),
        ),
    )]

    await db_session.refresh(request)
    await db_session.refresh(binding)
    await db_session.refresh(effect)
    assert request.reconcile_status == RECONCILE_STATUS_APPLIED
    assert binding.reconcile_status == RECONCILE_STATUS_APPLIED
    assert effect.status == EFFECT_STATUS_APPLIED
    assert effect.last_provider_outcome == "apply_present_applied"
    assert effect.last_provider_etag == "etag-after"

    run = await db_session.scalar(
        select(DataAccessEffectRun).where(DataAccessEffectRun.effect_id == effect.id),
    )
    assert run is not None
    assert run.mode == "apply_present"
    assert run.trace_id == f"data_access_apply_scheduled:{integration.id}:present"
    assert run.provider_outcome == "apply_present_applied"


def test_data_access_apply_handler_is_seeded_and_worker_registered() -> None:
    assert get_handler(DATA_ACCESS_BIGQUERY_APPLY_HANDLER) is not None

    server_root = Path(__file__).resolve().parents[2]
    worker_source = (server_root / "app" / "worker.py").read_text()
    migration_source = (
        server_root
        / "alembic"
        / "versions"
        / "341_seed_data_access_bigquery_apply_reconcile_job.py"
    ).read_text()

    assert "app.scheduled_jobs.data_access_handlers" in worker_source
    assert DATA_ACCESS_BIGQUERY_APPLY_HANDLER in migration_source


def test_apply_handler_failure_includes_exact_dataset_locator() -> None:
    failures: list[str] = []
    present = SimpleNamespace(
        config_error=None,
        selected_present_effect_count=1,
        error_count=1,
        conflict_count=0,
        results=(
            SimpleNamespace(
                outcome="apply_present_error",
                target_project_id="gcp-digital-bi",
                target_dataset_id="alibaba_cloud",
                target_table_id=None,
            ),
        ),
    )
    absent = SimpleNamespace(
        config_error=None,
        selected_absent_effect_count=0,
        error_count=0,
        conflict_count=0,
        results=(),
    )

    _collect_result_failures(
        failures,
        slug="dax-bigquery-prod",
        present=present,
        absent=absent,
    )

    assert failures == [
        "dax-bigquery-prod: 1 present apply error(s) at "
        "gcp-digital-bi.alibaba_cloud",
    ]


async def _build_present_effect_graph(
    db_session: Any,
    requester: Any,
    *,
    binding_status: str,
):
    """Build a DAX object graph (integration → entitlement → group → resource
    → request/item → binding → PRESENT effect) and return
    ``(integration, binding, effect)``. ``binding_status`` lets a test flip the
    binding to ``revoked_pending`` before the apply-present pass runs.
    """
    integration = Integration(
        kind=c.DOMAIN_DATA_ACCESS,
        provider=c.PROVIDER_BIGQUERY,
        name="DAX BigQuery",
        slug=f"dax-bq-guard-{uuid.uuid4().hex[:8]}",
        enabled=True,
        is_primary=False,
        config_public={"apply_mode_enabled": True},
    )
    entitlement = DataAccessEntitlement(
        domain=c.DOMAIN_DATA_ACCESS,
        provider=c.PROVIDER_BIGQUERY,
        resource_type=c.RESOURCE_TYPE_DATASET,
        principal_type=c.PRINCIPAL_TYPE_GOOGLE_GROUP,
        entitlement=c.ENTITLEMENT_READ,
        version=ENTITLEMENT_VERSION_V1,
        provider_member_prefix=c.BIGQUERY_GROUP_MEMBER_PREFIX,
        provider_role=c.BIGQUERY_DATA_VIEWER_ROLE,
        requestable=True,
        status=ENTITLEMENT_STATUS_ACTIVE,
    )
    group = EmailGroup(
        email=f"guard-readers-{uuid.uuid4().hex[:8]}@example.com",
        name="guard-readers",
        status="active",
        google_group_id=f"guard-google-group-{uuid.uuid4().hex[:8]}",
    )
    db_session.add_all([integration, entitlement, group])
    await db_session.flush()
    db_session.add(EmailGroupOwner(email_group_id=group.id, person_id=requester.id))
    resource = DataAccessResource(
        integration_id=integration.id,
        provider=c.PROVIDER_BIGQUERY,
        resource_type=c.RESOURCE_TYPE_DATASET,
        project_id="gcp-digital-bi",
        dataset_id="dwh_guard",
        display_name="Warehouse Guard",
        requestable=True,
        status=RESOURCE_STATUS_ACTIVE,
        owner_person_id=str(requester.id),
        owner_resolution_status=OWNER_RESOLUTION_RESOLVED,
        owner_source=OWNER_SOURCE_CATALOG,
        source_version="repo-test-v1",
        source_etag="repo-test-etag-v1",
    )
    db_session.add(resource)
    await db_session.flush()
    request = DataAccessRequest(
        requester_person_id=requester.id,
        principal_email_group_id=group.id,
        principal_google_group_id_snapshot=group.google_group_id,
        principal_email_snapshot=group.email,
        entitlement_id=entitlement.id,
        entitlement_snapshot=c.ENTITLEMENT_READ,
        entitlement_version_snapshot=ENTITLEMENT_VERSION_V1,
        approval_flow_key=c.APPROVAL_FLOW_DATA_ACCESS_GROUP_GRANT,
        feishu_instance_code=f"dax-guard-{uuid.uuid4().hex}",
        idempotency_key=f"dax-guard-{uuid.uuid4().hex}",
        business_reason="revoked-binding guard regression",
        status=REQUEST_STATUS_APPROVED,
        reconcile_status=RECONCILE_STATUS_PENDING,
        decided_at=datetime.now(UTC),
        trace_id="trace-dax-guard",
    )
    db_session.add(request)
    await db_session.flush()
    item = DataAccessRequestItem(
        request_id=request.id,
        resource_id=resource.id,
        line_no=1,
        resource_project_id_snapshot=resource.project_id,
        resource_dataset_id_snapshot=resource.dataset_id,
        resource_source_version_snapshot=resource.source_version,
        resource_source_etag_snapshot=resource.source_etag,
        status=REQUEST_ITEM_STATUS_APPROVED,
    )
    db_session.add(item)
    await db_session.flush()
    binding = DataAccessBinding(
        resource_id=resource.id,
        principal_email_group_id=group.id,
        principal_google_group_id_snapshot=group.google_group_id,
        principal_email_snapshot=group.email,
        entitlement_id=entitlement.id,
        entitlement_snapshot=c.ENTITLEMENT_READ,
        entitlement_version_snapshot=ENTITLEMENT_VERSION_V1,
        source_request_id=request.id,
        source_request_item_id=item.id,
        status=binding_status,
        reconcile_status=RECONCILE_STATUS_PENDING,
    )
    db_session.add(binding)
    await db_session.flush()
    effect = DataAccessCompiledEffect(
        binding_id=binding.id,
        resource_id=resource.id,
        principal_email_group_id=group.id,
        entitlement_id=entitlement.id,
        provider=c.PROVIDER_BIGQUERY,
        resource_type=c.RESOURCE_TYPE_DATASET,
        target_project_id=resource.project_id,
        target_dataset_id=resource.dataset_id,
        provider_member=f"{c.BIGQUERY_GROUP_MEMBER_PREFIX}{group.email}",
        provider_role=c.BIGQUERY_DATA_VIEWER_ROLE,
        desired_state=EFFECT_DESIRED_PRESENT,
        entitlement_version_snapshot=ENTITLEMENT_VERSION_V1,
        resource_source_version_snapshot=resource.source_version,
        resource_source_etag_snapshot=resource.source_etag,
        compiler_version="dax-compiler-v1",
        managed_reason=EFFECT_MANAGED_REASON_GRANT_REQUEST,
        status=EFFECT_STATUS_PENDING_RECONCILE,
    )
    db_session.add(effect)
    await db_session.flush()
    return integration, binding, effect


async def test_apply_present_skips_effect_whose_binding_is_revoked(
    db_session,
    employee_user,
) -> None:
    """Regression (#1643): a binding revoked while its PRESENT effect is still
    pending must NOT be re-written into the dataset IAM. apply-present has to
    skip effects whose ``binding.status`` is no longer ACTIVE — symmetric with
    the absent-side guard — or the next apply pass resurrects revoked access.
    """
    from app.domains.data_access.reconciler import (
        apply_present_reconcile_compiled_effects,
    )

    requester, _ = employee_user
    integration, binding, effect = await _build_present_effect_graph(
        db_session, requester, binding_status=BINDING_STATUS_REVOKED_PENDING,
    )

    fake_driver = _FakeApplyDriver()
    result = await apply_present_reconcile_compiled_effects(
        db_session,
        integration_id=integration.id,
        driver=fake_driver,
    )

    # The revoked-binding PRESENT effect is filtered out before the gate — it
    # is neither selected nor written to the provider.
    assert result.selected_present_effect_count == 0
    assert result.provider_write_count == 0
    assert fake_driver.get_calls == []
    assert fake_driver.set_calls == []

    # And the effect is left untouched (still pending) rather than applied.
    await db_session.refresh(effect)
    assert effect.status == EFFECT_STATUS_PENDING_RECONCILE


async def test_apply_present_applies_effect_when_binding_active(
    db_session,
    employee_user,
) -> None:
    """Control for #1643: with an ACTIVE binding the same graph DOES apply,
    proving the guard only suppresses revoked bindings."""
    from app.domains.data_access.reconciler import (
        apply_present_reconcile_compiled_effects,
    )

    requester, _ = employee_user
    integration, binding, effect = await _build_present_effect_graph(
        db_session, requester, binding_status=BINDING_STATUS_ACTIVE,
    )

    fake_driver = _FakeApplyDriver()
    result = await apply_present_reconcile_compiled_effects(
        db_session,
        integration_id=integration.id,
        driver=fake_driver,
    )

    assert result.selected_present_effect_count == 1
    assert result.provider_write_count == 1
    assert len(fake_driver.set_calls) == 1
    await db_session.refresh(effect)
    assert effect.status == EFFECT_STATUS_APPLIED