from __future__ import annotations

from collections.abc import Callable

from sqlalchemy import func, inspect, select

from govoplan_core.core.modules import MigrationRetirementPlan, ModuleUninstallGuardResult


def persistent_table_uninstall_guard(
    *models: object,
    label: str | None = None,
) -> Callable[[object | None, str], tuple[ModuleUninstallGuardResult, ...]]:
    def guard(session: object | None, module_id: str) -> tuple[ModuleUninstallGuardResult, ...]:
        if session is None or not hasattr(session, "execute") or not hasattr(session, "get_bind"):
            return (
                ModuleUninstallGuardResult(
                    "warning",
                    "uninstall_guard_no_session",
                    "Persistent-data uninstall guard could not inspect the database without a session.",
                ),
            )
        bind = session.get_bind()  # type: ignore[attr-defined]
        inspector = inspect(bind)
        populated_tables: list[str] = []
        for model in models:
            table = getattr(model, "__table__", None)
            if table is None or not inspector.has_table(table.name):
                continue
            row_count = int(session.execute(select(func.count()).select_from(table)).scalar_one())  # type: ignore[attr-defined]
            if row_count:
                populated_tables.append(f"{table.name} ({row_count})")
        if not populated_tables:
            return ()
        table_summary = ", ".join(populated_tables[:8])
        if len(populated_tables) > 8:
            table_summary += f", and {len(populated_tables) - 8} more"
        return (
            ModuleUninstallGuardResult(
                "warning",
                "persistent_data_present",
                f"{label or module_id} owns persistent data: {table_summary}. Non-destructive uninstall leaves this data dormant until the module is reinstalled or an explicit retirement provider handles it.",
            ),
        )

    return guard


def drop_table_retirement_provider(
    *models: object,
    label: str | None = None,
) -> Callable[[object | None, str], MigrationRetirementPlan]:
    def provider(session: object | None, module_id: str) -> MigrationRetirementPlan:
        module_label = label or module_id
        tables = tuple(_model_tables(models))
        table_names = tuple(table.name for table in tables)
        if session is None or not hasattr(session, "get_bind"):
            return MigrationRetirementPlan(
                supported=False,
                summary=f"{module_label} table retirement could not inspect the database without a session.",
                blocking_reason="No database session is available.",
            )
        if not tables:
            return MigrationRetirementPlan(
                supported=True,
                summary=f"{module_label} declares no module-owned tables to retire.",
                destroy_data_supported=True,
                destroy_data_summary=f"{module_label} has no module-owned tables to drop.",
                destroy_data_executor=lambda _session, _module_id: None,
            )
        bind = session.get_bind()  # type: ignore[attr-defined]
        inspector = inspect(bind)
        existing = tuple(table for table in tables if inspector.has_table(table.name))
        existing_names = tuple(table.name for table in existing)
        missing_names = tuple(name for name in table_names if name not in existing_names)
        summary = f"{module_label} can retire {len(existing_names)} table(s): {', '.join(existing_names) if existing_names else 'none currently present'}."
        warnings: list[str] = []
        if missing_names:
            warnings.append("Tables not present and therefore skipped: " + ", ".join(missing_names))

        def executor(execute_session: object, _module_id: str) -> None:
            if not hasattr(execute_session, "connection"):
                raise RuntimeError("No database session is available for destructive table retirement.")
            # Enlist schema retirement in the caller's active transaction.  An
            # Engine returned by Session.get_bind() may acquire a second
            # connection, separating the DROP from module-specific secret
            # scrubbing/audit writes and deadlocking on their uncommitted locks.
            execute_connection = execute_session.connection()  # type: ignore[attr-defined]
            live_inspector = inspect(execute_connection)
            live_tables = [table for table in tables if live_inspector.has_table(table.name)]
            if not live_tables:
                return
            metadata = live_tables[0].metadata
            metadata.drop_all(bind=execute_connection, tables=live_tables, checkfirst=True)

        return MigrationRetirementPlan(
            supported=True,
            summary=summary,
            warnings=tuple(warnings),
            destroy_data_supported=True,
            destroy_data_summary=(
                f"Destroying data will drop {module_label} table(s): "
                + (", ".join(existing_names) if existing_names else "none currently present")
                + "."
            ),
            destroy_data_warnings=(
                "This permanently deletes module-owned rows unless the installer rollback restores the database snapshot.",
            ),
            destroy_data_executor=executor,
        )

    return provider


def _model_tables(models: tuple[object, ...]) -> tuple[object, ...]:
    tables: list[object] = []
    for model in models:
        table = getattr(model, "__table__", None)
        if table is not None:
            tables.append(table)
    return tuple(tables)
