"""Scheduled handlers for data-access provider reconciliation."""
from __future__ import annotations

import logging
from collections.abc import Mapping
from typing import Any

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.domains.data_access import constants as c
from app.domains.data_access.apply_gate import data_access_apply_gate_config
from app.domains.data_access.reconciler import (
    DataAccessApplyAbsentReconcileResult,
    DataAccessApplyPresentReconcileResult,
    apply_absent_reconcile_compiled_effects,
    apply_present_reconcile_compiled_effects,
)
from app.domains.data_access.service import DataAccessService
from app.integrations.models import Integration
from app.scheduled_jobs.registry import register_scheduled_job

logger = logging.getLogger(__name__)

DATA_ACCESS_BIGQUERY_APPLY_HANDLER = "data_access.bigquery_apply_reconcile"


@register_scheduled_job(
    key=DATA_ACCESS_BIGQUERY_APPLY_HANDLER,
    display_name="DAX BigQuery 权限变更执行",
    description=(
        "定时执行已批准 DAX 请求编译出的 BigQuery dataset IAM present/absent "
        "effects。只处理 data_access/bigquery 集成,并继续受 apply_mode_enabled "
        "门控保护。"
    ),
)
async def run_data_access_bigquery_apply_reconcile(
    db: AsyncSession,
    config: dict[str, Any],
) -> dict[str, Any]:
    del config

    personal_identity_revoke = await DataAccessService().revoke_stale_personal_bindings(
        db,
        trace_id="data_access_apply_scheduled:personal_identity_revoke",
    )
    # Persist the semantic absent effects before the later provider phases.
    # BigQuery remains untouched until apply-absent receives its normal gate.
    await db.commit()

    integrations = list(
        (
            await db.scalars(
                select(Integration)
                .where(
                    Integration.kind == c.DOMAIN_DATA_ACCESS,
                    Integration.provider == c.PROVIDER_BIGQUERY,
                    Integration.enabled.is_(True),
                )
                .order_by(Integration.slug, Integration.created_at)
            )
        ).all()
    )

    summary: dict[str, Any] = {
        "integration_count": len(integrations),
        "apply_enabled_integration_count": 0,
        "apply_disabled_integration_count": 0,
        "invalid_config_integration_count": 0,
        "personal_identity_revoke": {
            "revoked_pending_count": personal_identity_revoke.revoked_pending_count,
            "effect_count": personal_identity_revoke.effect_count,
        },
        "present": _empty_present_summary(),
        "absent": _empty_absent_summary(),
        "integrations": [],
    }
    failures: list[str] = []

    for integration in integrations:
        config_public = integration.config_public
        if config_public is not None and not isinstance(config_public, Mapping):
            message = (
                f"{integration.slug}: config_public must be an object for "
                "data_access/bigquery apply"
            )
            failures.append(message)
            summary["invalid_config_integration_count"] += 1
            summary["integrations"].append({
                "integration_id": str(integration.id),
                "slug": integration.slug,
                "status": "invalid_config",
                "error": message,
            })
            continue

        try:
            gate_config = data_access_apply_gate_config(config_public)
        except ValueError as exc:
            message = f"{integration.slug}: {exc}"
            failures.append(message)
            summary["invalid_config_integration_count"] += 1
            summary["integrations"].append({
                "integration_id": str(integration.id),
                "slug": integration.slug,
                "status": "invalid_config",
                "error": message,
            })
            continue

        if not gate_config.apply_mode_enabled:
            summary["apply_disabled_integration_count"] += 1
            summary["integrations"].append({
                "integration_id": str(integration.id),
                "slug": integration.slug,
                "status": "apply_mode_disabled",
            })
            continue

        summary["apply_enabled_integration_count"] += 1
        trace_prefix = f"data_access_apply_scheduled:{integration.id}"
        # Provider writes cannot be part of an all-or-nothing DB transaction:
        # if a later phase fails and the scheduler rolls back this handler
        # session, BigQuery would already be changed while Helix still says
        # pending. Commit each apply phase after its existing reconciler has
        # written effect-run evidence and rollups.
        present = await apply_present_reconcile_compiled_effects(
            db,
            integration_id=integration.id,
            trace_id=f"{trace_prefix}:present",
        )
        await db.commit()
        absent = await apply_absent_reconcile_compiled_effects(
            db,
            integration_id=integration.id,
            trace_id=f"{trace_prefix}:absent",
        )
        await db.commit()
        present_summary = _present_summary(present)
        absent_summary = _absent_summary(absent)
        _merge_counts(summary["present"], present_summary)
        _merge_counts(summary["absent"], absent_summary)
        summary["integrations"].append({
            "integration_id": str(integration.id),
            "slug": integration.slug,
            "status": "processed",
            "present": present_summary,
            "absent": absent_summary,
        })

        _collect_result_failures(
            failures,
            slug=integration.slug,
            present=present,
            absent=absent,
        )

    if failures:
        compact = "; ".join(failures[:5])
        if len(failures) > 5:
            compact = f"{compact}; ... (+{len(failures) - 5} more)"
        raise RuntimeError(f"{DATA_ACCESS_BIGQUERY_APPLY_HANDLER}: {compact}")

    logger.info("data-access BigQuery apply reconcile complete: %s", summary)
    return summary


def _empty_present_summary() -> dict[str, int]:
    return {
        "selected_effect_count": 0,
        "selected_present_effect_count": 0,
        "ignored_non_present_effect_count": 0,
        "allowed_effect_count": 0,
        "blocked_effect_count": 0,
        "provider_write_count": 0,
        "applied_count": 0,
        "already_applied_count": 0,
        "conflict_count": 0,
        "error_count": 0,
    }


def _empty_absent_summary() -> dict[str, int]:
    return {
        "selected_effect_count": 0,
        "selected_absent_effect_count": 0,
        "ignored_non_absent_effect_count": 0,
        "ignored_non_revoked_binding_effect_count": 0,
        "allowed_effect_count": 0,
        "blocked_effect_count": 0,
        "provider_write_count": 0,
        "removed_count": 0,
        "already_absent_count": 0,
        "conflict_count": 0,
        "error_count": 0,
    }


def _present_summary(
    result: DataAccessApplyPresentReconcileResult,
) -> dict[str, int | str | bool | None]:
    return {
        "run_id": str(result.run_id),
        "apply_mode_enabled": result.apply_mode_enabled,
        "selected_effect_count": result.selected_effect_count,
        "selected_present_effect_count": result.selected_present_effect_count,
        "ignored_non_present_effect_count": result.ignored_non_present_effect_count,
        "allowed_effect_count": result.allowed_effect_count,
        "blocked_effect_count": result.blocked_effect_count,
        "provider_write_count": result.provider_write_count,
        "applied_count": result.applied_count,
        "already_applied_count": result.already_applied_count,
        "conflict_count": result.conflict_count,
        "error_count": result.error_count,
        "config_error": result.config_error,
    }


def _absent_summary(
    result: DataAccessApplyAbsentReconcileResult,
) -> dict[str, int | str | bool | None]:
    return {
        "run_id": str(result.run_id),
        "apply_mode_enabled": result.apply_mode_enabled,
        "selected_effect_count": result.selected_effect_count,
        "selected_absent_effect_count": result.selected_absent_effect_count,
        "ignored_non_absent_effect_count": result.ignored_non_absent_effect_count,
        "ignored_non_revoked_binding_effect_count": (
            result.ignored_non_revoked_binding_effect_count
        ),
        "allowed_effect_count": result.allowed_effect_count,
        "blocked_effect_count": result.blocked_effect_count,
        "provider_write_count": result.provider_write_count,
        "removed_count": result.removed_count,
        "already_absent_count": result.already_absent_count,
        "conflict_count": result.conflict_count,
        "error_count": result.error_count,
        "config_error": result.config_error,
    }


def _merge_counts(
    target: dict[str, int],
    source: dict[str, int | str | bool | None],
) -> None:
    for key, value in source.items():
        if key in target and isinstance(value, int) and not isinstance(value, bool):
            target[key] += value


def _collect_result_failures(
    failures: list[str],
    *,
    slug: str,
    present: DataAccessApplyPresentReconcileResult,
    absent: DataAccessApplyAbsentReconcileResult,
) -> None:
    if present.config_error and present.selected_present_effect_count:
        failures.append(f"{slug}: present config error: {present.config_error}")
    if absent.config_error and absent.selected_absent_effect_count:
        failures.append(f"{slug}: absent config error: {absent.config_error}")
    if present.error_count:
        failures.append(
            f"{slug}: {present.error_count} present apply error(s)"
            f"{_result_locator_suffix(present.results, outcome='apply_present_error')}"
        )
    if present.conflict_count:
        failures.append(
            f"{slug}: {present.conflict_count} present conflict(s)"
            f"{_result_locator_suffix(present.results, outcome='apply_present_conflict')}"
        )
    if absent.error_count:
        failures.append(
            f"{slug}: {absent.error_count} absent apply error(s)"
            f"{_result_locator_suffix(absent.results, outcome='apply_absent_error')}"
        )
    if absent.conflict_count:
        failures.append(
            f"{slug}: {absent.conflict_count} absent conflict(s)"
            f"{_result_locator_suffix(absent.results, outcome='apply_absent_conflict')}"
        )


def _result_locator_suffix(
    results: tuple[Any, ...],
    *,
    outcome: str,
) -> str:
    locators = sorted({
        ".".join(
            part
            for part in (
                row.target_project_id,
                row.target_dataset_id,
                row.target_table_id,
            )
            if part
        )
        for row in results
        if row.outcome == outcome
    })
    if not locators:
        return ""
    visible = ", ".join(locators[:3])
    if len(locators) > 3:
        visible = f"{visible}, ... (+{len(locators) - 3} more)"
    return f" at {visible}"


__all__ = [
    "DATA_ACCESS_BIGQUERY_APPLY_HANDLER",
    "run_data_access_bigquery_apply_reconcile",
]