from __future__ import annotations

from collections.abc import Iterable, Mapping
from dataclasses import dataclass, replace
import json
import logging
import os
from pathlib import Path
import re
import sysconfig
from typing import Any

from alembic import command
from alembic.config import Config
from alembic.runtime.migration import MigrationContext
from alembic.script import ScriptDirectory
from sqlalchemy import create_engine, inspect, text

from govoplan_core.core.migrations import MigrationMetadataPlan, migration_metadata_plan
from govoplan_core.core import change_sequence as core_change_sequence_models  # noqa: F401 - populate core metadata
from govoplan_core.core import recovery as core_recovery_models  # noqa: F401 - populate core metadata
from govoplan_core.core import runtime_coordination as core_runtime_models  # noqa: F401 - populate core metadata
from govoplan_core.security import credential_envelopes as core_credential_models  # noqa: F401 - populate core metadata
from govoplan_core.core.change_sequence import ChangeSequenceEntry, ChangeSequenceRetentionFloor
from govoplan_core.core.module_management import load_startup_enabled_modules, startup_candidate_module_ids
from govoplan_core.core.modules import ModuleMigrationTaskContext, ModuleMigrationTaskResult
from govoplan_core.db.session import configure_database, get_database
from govoplan_core.server.default_config import get_server_config
from govoplan_core.server.config import ManifestFactory
from govoplan_core.server.registry import available_module_manifests, build_platform_registry
from govoplan_core.settings import settings

logger = logging.getLogger(__name__)

# Historic development databases could be created partly through Alembic and
# partly through Base.metadata.create_all(). In that state Alembic still says
# "2c..." while the 3d/4e file-storage tables already exist, so a normal
# upgrade attempts to create file_blobs again. This reconciliation is kept
# deliberately narrow and only advances the marker when the complete expected
# schema for the skipped revisions is already present.
REVISION_AUTH_RBAC = "2c3d4e5f6a7b"
REVISION_FILE_STORAGE = "3d4e5f6a7b8c"
REVISION_FILE_FOLDERS = "4e5f6a7b8c9d"
REVISION_NAMESPACE_PLATFORM_TABLES = "2e3f4a5b6c7d"
REVISION_CORE_CHANGE_SEQUENCE = "3f4a5b6c7d8e"
REVISION_HIERARCHICAL_SETTINGS = "f5a6b7c8d9e0"
LEGACY_SCOPE_TABLE = "tenancy_tenants"
CORE_SCOPE_TABLE = "core_scopes"
PRE_MIGRATION_TASK_PHASES = ("pre_migration_check", "pre_migration_prepare")
POST_MIGRATION_TASK_PHASES = ("post_migration_backfill", "post_migration_verify")
MIGRATION_TASK_PHASES = (*PRE_MIGRATION_TASK_PHASES, *POST_MIGRATION_TASK_PHASES)
MIGRATION_TRACK_RELEASE = "release"
MIGRATION_TRACK_DEV = "dev"
MIGRATION_TRACKS = (MIGRATION_TRACK_RELEASE, MIGRATION_TRACK_DEV)
_SQL_IDENTIFIER_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")

_NAMESPACE_TABLE_RENAMES = (
    ("tenants", "tenancy_tenants"),
    ("accounts", "access_accounts"),
    ("users", "access_users"),
    ("groups", "access_groups"),
    ("roles", "access_roles"),
    ("system_role_assignments", "access_system_role_assignments"),
    ("user_group_memberships", "access_user_group_memberships"),
    ("user_role_assignments", "access_user_role_assignments"),
    ("group_role_assignments", "access_group_role_assignments"),
    ("api_keys", "access_api_keys"),
    ("auth_sessions", "access_auth_sessions"),
    ("system_settings", "core_system_settings"),
    ("governance_templates", "admin_governance_templates"),
    ("governance_template_assignments", "admin_governance_template_assignments"),
)

_FILE_STORAGE_TABLES = {
    "file_blobs",
    "file_assets",
    "file_versions",
    "file_shares",
    "campaign_attachment_uses",
}
_FILE_FOLDER_TABLES = {"file_folders"}

_FILE_STORAGE_COLUMNS = {
    "file_blobs": {
        "id",
        "tenant_id",
        "storage_backend",
        "storage_key",
        "checksum_sha256",
        "size_bytes",
    },
    "file_assets": {
        "id",
        "tenant_id",
        "owner_type",
        "display_path",
        "filename",
        "current_version_id",
    },
    "file_versions": {
        "id",
        "file_asset_id",
        "blob_id",
        "version_number",
        "checksum_sha256",
    },
    "file_shares": {"id", "file_asset_id", "target_type", "target_id", "permission"},
    "campaign_attachment_uses": {
        "id",
        "campaign_id",
        "campaign_version_id",
        "file_asset_id",
        "file_version_id",
        "file_blob_id",
    },
    "file_folders": {"id", "tenant_id", "owner_type", "path"},
}

_CREATE_ALL_THROUGH_HIERARCHICAL_TABLES = {
    "access_accounts",
    "access_auth_sessions",
    "audit_log",
    "campaign_versions",
    "campaign_jobs",
    "send_attempts",
    "access_system_role_assignments",
    "core_system_settings",
    "admin_governance_templates",
    "admin_governance_template_assignments",
    "mail_server_profiles",
}

_CREATE_ALL_THROUGH_HIERARCHICAL_COLUMNS = {
    "access_auth_sessions": {"account_id", "csrf_token_hash"},
    "audit_log": {"scope", "tenant_id", "user_id", "api_key_id"},
    "campaign_versions": {
        "workflow_state",
        "current_flow",
        "user_lock_state",
        "user_locked_at",
        "user_locked_by_user_id",
        "execution_snapshot",
        "execution_snapshot_hash",
        "execution_snapshot_at",
    },
    "campaign_jobs": {"claimed_at", "claim_token", "smtp_started_at", "outcome_unknown_at", "eml_sha256"},
    "send_attempts": {"status", "claim_token"},
    "access_users": {"account_id", "settings", "mail_profile_policy"},
    "access_groups": {"system_template_id", "system_required", "settings", "mail_profile_policy"},
    "access_roles": {"system_template_id", "system_required"},
    "campaigns": {"settings", "mail_profile_policy"},
    "mail_server_profiles": {"scope_type", "scope_id"},
    "core_system_settings": {"settings", "allow_tenant_custom_groups", "allow_tenant_custom_roles", "allow_tenant_api_keys"},
}

_SCOPE_TABLE_COLUMNS = {"settings", "allow_custom_groups", "allow_custom_roles", "allow_api_keys"}

_CREATE_ALL_THROUGH_HIERARCHICAL_INDEXES = {
    "file_folders": {"uq_file_folders_active_user_path", "uq_file_folders_active_group_path"},
    "campaign_versions": {
        "ix_campaign_versions_user_lock_state",
        "ix_campaign_versions_user_locked_by_user_id",
        "ix_campaign_versions_execution_snapshot_hash",
    },
    "campaign_jobs": {"ix_campaign_jobs_claim_token", "ix_campaign_jobs_eml_sha256"},
    "send_attempts": {"ix_send_attempts_status", "ix_send_attempts_claim_token"},
    "audit_log": {"ix_audit_log_scope_created_at", "ix_audit_log_tenant_scope_created_at"},
    "access_auth_sessions": {"ix_access_auth_sessions_account_id"},
    "access_users": {"ix_access_users_account_id"},
    "access_groups": {"ix_access_groups_system_template_id"},
    "access_roles": {"ix_access_roles_system_template_id"},
    "mail_server_profiles": {"ix_mail_server_profiles_scope_type", "ix_mail_server_profiles_scope_id", "ix_mail_server_profiles_scope"},
}


@dataclass(frozen=True, slots=True)
class _LegacyCreateAllSchemaState:
    current: str | None
    has_no_revision: bool
    has_file_storage: bool
    has_file_folders: bool
    has_create_all_hierarchical_schema: bool


@dataclass(frozen=True, slots=True)
class MigrationResult:
    previous_revision: str | None
    reconciled_revision: str | None
    current_revision: str | None


class ModuleMigrationTaskExecutionError(RuntimeError):
    def __init__(self, message: str, *, records: tuple[dict[str, object], ...]) -> None:
        super().__init__(message)
        self.records = records


def registered_module_migration_plan(
    database_url: str | None = None,
    *,
    enabled_modules: tuple[str, ...] | list[str] | None = None,
    manifest_factories: tuple[ManifestFactory, ...] = (),
    migration_track: str | None = None,
) -> MigrationMetadataPlan:
    plan = migration_metadata_plan(_registered_module_registry(
        database_url=database_url,
        enabled_modules=enabled_modules,
        manifest_factories=manifest_factories,
    ))
    return _migration_plan_for_track(plan, _normalize_migration_track(migration_track))


def _registered_module_registry(
    *,
    database_url: str | None = None,
    enabled_modules: tuple[str, ...] | list[str] | None = None,
    manifest_factories: tuple[ManifestFactory, ...] = (),
):
    server_config = get_server_config()
    active_database_url = database_url or settings.database_url
    if active_database_url:
        # Registry planning may target a different database than the currently
        # configured handle. The global handle is replaced either way, so
        # dispose its superseded pool instead of leaking open DBAPI connections.
        configure_database(active_database_url, dispose_previous=True)
    active_manifest_factories = manifest_factories or tuple(server_config.manifest_factories)
    raw_enabled_modules = tuple(enabled_modules) if enabled_modules is not None else load_startup_enabled_modules(server_config.enabled_modules)
    candidate_modules = startup_candidate_module_ids(server_config.enabled_modules, raw_enabled_modules)
    available_modules = available_module_manifests(
        active_manifest_factories,
        enabled_modules=candidate_modules,
        ignore_load_errors=True,
    )
    active_enabled_modules = (
        tuple(enabled_modules)
        if enabled_modules is not None
        else load_startup_enabled_modules(server_config.enabled_modules, available=available_modules)
    )
    registry = build_platform_registry(
        active_enabled_modules,
        manifest_factories=active_manifest_factories,
    )
    return registry


def run_registered_module_migration_tasks(
    *,
    database_url: str | None = None,
    enabled_modules: tuple[str, ...] | list[str] | None = None,
    migration_module_order: tuple[str, ...] | list[str] | None = None,
    phases: tuple[str, ...] | list[str] = MIGRATION_TASK_PHASES,
    dry_run: bool = False,
    manifest_factories: tuple[ManifestFactory, ...] = (),
) -> tuple[dict[str, object], ...]:
    active_phases = _normalized_migration_task_phases(phases)
    url = database_url or settings.database_url
    registry = _registered_module_registry(
        database_url=url,
        enabled_modules=enabled_modules,
        manifest_factories=manifest_factories,
    )
    manifests = {manifest.id: manifest for manifest in registry.manifests()}
    ordered_ids = _ordered_migration_task_module_ids(migration_module_order, manifests)
    records: list[dict[str, object]] = []
    database = get_database()
    with database.SessionLocal() as session:
        for phase in active_phases:
            for manifest, task in _iter_phase_migration_tasks(phase, ordered_ids=ordered_ids, manifests=manifests):
                _run_registered_module_migration_task(
                    session,
                    manifest=manifest,
                    task=task,
                    database_url=url,
                    dry_run=dry_run,
                    records=records,
                )
    return tuple(records)


def _normalized_migration_task_phases(phases: tuple[str, ...] | list[str]) -> tuple[str, ...]:
    active_phases = tuple(dict.fromkeys(str(item).strip() for item in phases if str(item).strip()))
    invalid_phases = tuple(phase for phase in active_phases if phase not in MIGRATION_TASK_PHASES)
    if invalid_phases:
        raise ValueError("Unsupported module migration task phase(s): " + ", ".join(invalid_phases))
    return active_phases


def _ordered_migration_task_module_ids(
    migration_module_order: tuple[str, ...] | list[str] | None,
    manifests: Mapping[str, object],
) -> tuple[str, ...]:
    return tuple(dict.fromkeys([
        *(str(item).strip() for item in (migration_module_order or ()) if str(item).strip()),
        *manifests.keys(),
    ]))


def _iter_phase_migration_tasks(
    phase: str,
    *,
    ordered_ids: tuple[str, ...],
    manifests: Mapping[str, object],
) -> Iterable[tuple[object, object]]:
    for module_id in ordered_ids:
        manifest = manifests.get(module_id)
        migration_spec = getattr(manifest, "migration_spec", None)
        if manifest is None or migration_spec is None:
            continue
        for task in migration_spec.migration_tasks:
            if task.phase == phase:
                yield manifest, task


def _run_registered_module_migration_task(
    session: object,
    *,
    manifest: object,
    task: object,
    database_url: str,
    dry_run: bool,
    records: list[dict[str, object]],
) -> None:
    record = _migration_task_record(manifest, task, dry_run=dry_run)
    _validate_migration_task_can_run(manifest, task, record, records, dry_run=dry_run)
    if dry_run:
        record.update({"status": "skipped", "message": "Dry run; executor was not called."})
        records.append(record)
        return
    normalized = _execute_module_migration_task(
        session,
        manifest=manifest,
        task=task,
        record=record,
        records=records,
        database_url=database_url,
        dry_run=dry_run,
    )
    record.update({
        "status": normalized.status,
        "message": normalized.message,
        "details": _jsonable_migration_task_details(normalized.details),
    })
    records.append(record)
    if normalized.status == "blocked":
        session.rollback()
        raise ModuleMigrationTaskExecutionError(
            normalized.message or f"Module migration task {manifest.id}/{task.task_id} blocked migration.",
            records=tuple(records),
        )
    session.commit()


def _migration_task_record(manifest: object, task: object, *, dry_run: bool) -> dict[str, object]:
    record: dict[str, object] = {
        "module_id": manifest.id,
        "task_id": task.task_id,
        "phase": task.phase,
        "summary": task.summary,
        "task_version": task.task_version,
        "safety": task.safety,
        "idempotent": task.idempotent,
        "dry_run": dry_run,
    }
    if task.timeout_seconds is not None:
        record["timeout_seconds"] = task.timeout_seconds
    return record


def _validate_migration_task_can_run(
    manifest: object,
    task: object,
    record: dict[str, object],
    records: list[dict[str, object]],
    *,
    dry_run: bool,
) -> None:
    if not task.idempotent:
        _block_migration_task(
            manifest,
            task,
            record,
            records,
            message="Task is not idempotent.",
            error=f"Module migration task {manifest.id}/{task.task_id} is not idempotent.",
        )
    if not dry_run and task.executor is None:
        _block_migration_task(
            manifest,
            task,
            record,
            records,
            message="Task has no executor.",
            error=f"Module migration task {manifest.id}/{task.task_id} has no executor.",
        )


def _block_migration_task(
    manifest: object,
    task: object,
    record: dict[str, object],
    records: list[dict[str, object]],
    *,
    message: str,
    error: str,
) -> None:
    record.update({"status": "blocked", "message": message})
    records.append(record)
    raise ModuleMigrationTaskExecutionError(
        error,
        records=tuple(records),
    )


def _execute_module_migration_task(
    session: object,
    *,
    manifest: object,
    task: object,
    record: dict[str, object],
    records: list[dict[str, object]],
    database_url: str,
    dry_run: bool,
) -> ModuleMigrationTaskResult:
    context = ModuleMigrationTaskContext(
        module_id=manifest.id,
        task_id=task.task_id,
        phase=task.phase,
        database_url=database_url,
        target_version=manifest.version,
        session=session,
        dry_run=dry_run,
        metadata=task.metadata,
    )
    try:
        return _normalize_migration_task_result(task.executor(context))
    except Exception as exc:
        session.rollback()
        record.update({
            "status": "blocked",
            "message": f"{type(exc).__name__}: {exc}",
        })
        records.append(record)
        raise ModuleMigrationTaskExecutionError(
            f"Module migration task {manifest.id}/{task.task_id} failed.",
            records=tuple(records),
        ) from exc


def _normalize_migration_task_result(result: ModuleMigrationTaskResult | None) -> ModuleMigrationTaskResult:
    if result is None:
        return ModuleMigrationTaskResult()
    if not isinstance(result, ModuleMigrationTaskResult):
        raise TypeError("Module migration task executors must return ModuleMigrationTaskResult or None.")
    if result.status not in {"ok", "warning", "blocked", "skipped"}:
        raise ValueError(f"Unsupported module migration task status: {result.status!r}")
    return result


def _jsonable_migration_task_details(value: Mapping[str, Any]) -> Mapping[str, Any] | str:
    try:
        json.dumps(value)
    except (TypeError, ValueError):
        return str(dict(value))
    return value


def _repo_root() -> Path:
    packaged_root = Path(__file__).resolve().parents[3]
    installed_runtime_root = Path(sysconfig.get_path("data")) / "govoplan_core_runtime"
    configured = os.environ.get("GOVOPLAN_CORE_SOURCE_ROOT")
    candidates = [
        Path(configured).expanduser() if configured else None,
        Path.cwd(),
        packaged_root,
        installed_runtime_root,
    ]
    for candidate in candidates:
        if candidate is None:
            continue
        resolved = candidate.resolve()
        if (resolved / "alembic.ini").exists() and (resolved / "alembic").is_dir():
            return resolved
    return packaged_root


def _normalize_migration_track(value: str | None = None) -> str:
    raw_value = value if value is not None else settings.migration_track
    track = str(raw_value or MIGRATION_TRACK_RELEASE).strip().lower()
    aliases = {
        "baseline": MIGRATION_TRACK_RELEASE,
        "prod": MIGRATION_TRACK_RELEASE,
        "production": MIGRATION_TRACK_RELEASE,
        "releases": MIGRATION_TRACK_RELEASE,
        "detailed": MIGRATION_TRACK_DEV,
        "development": MIGRATION_TRACK_DEV,
    }
    track = aliases.get(track, track)
    if track not in MIGRATION_TRACKS:
        raise ValueError(
            "Unsupported migration track "
            f"{raw_value!r}; expected one of: {', '.join(MIGRATION_TRACKS)}"
        )
    return track


def _core_version_location(root: Path, migration_track: str) -> Path:
    if migration_track == MIGRATION_TRACK_DEV:
        dev_versions = root / "alembic" / "dev_versions"
        if dev_versions.is_dir():
            return dev_versions
    return root / "alembic" / "versions"


def _script_location_for_track(location: str | None, migration_track: str) -> str | None:
    if not location or migration_track != MIGRATION_TRACK_DEV:
        return location
    path = Path(location)
    if path.name != "versions":
        return location
    dev_location = path.with_name("dev_versions")
    if dev_location.is_dir():
        return str(dev_location)
    return location


def _migration_plan_for_track(plan: MigrationMetadataPlan, migration_track: str) -> MigrationMetadataPlan:
    if migration_track == MIGRATION_TRACK_RELEASE:
        return plan
    script_locations = tuple(
        location
        for location in (
            _script_location_for_track(location, migration_track)
            for location in plan.script_locations
        )
        if location
    )
    modules = tuple(
        replace(
            item,
            script_location=_script_location_for_track(item.script_location, migration_track),
        )
        for item in plan.modules
    )
    return replace(plan, script_locations=script_locations, modules=modules)


def alembic_config(
    *,
    database_url: str | None = None,
    enabled_modules: tuple[str, ...] | list[str] | None = None,
    manifest_factories: tuple[ManifestFactory, ...] = (),
    migration_track: str | None = None,
) -> Config:
    root = _repo_root()
    active_migration_track = _normalize_migration_track(migration_track)
    config = Config(str(root / "alembic.ini"))
    config.set_main_option("script_location", str(root / "alembic"))

    active_database_url = database_url or settings.database_url
    plan = registered_module_migration_plan(
        active_database_url,
        enabled_modules=enabled_modules,
        manifest_factories=manifest_factories,
        migration_track=active_migration_track,
    )
    version_locations = [str(_core_version_location(root, active_migration_track))]
    version_locations.extend(location for location in plan.script_locations if location)
    config.set_main_option("version_locations", os.pathsep.join(dict.fromkeys(version_locations)))
    config.set_main_option("path_separator", "os")

    config.attributes["database_url"] = active_database_url
    config.attributes["migration_track"] = active_migration_track
    if enabled_modules is not None:
        config.attributes["enabled_modules"] = tuple(enabled_modules)
    if manifest_factories:
        config.attributes["manifest_factories"] = tuple(manifest_factories)
    return config


def database_revision(database_url: str | None = None) -> str | None:
    url = database_url or settings.database_url
    engine = create_engine(url)
    try:
        with engine.connect() as connection:
            heads = MigrationContext.configure(connection).get_current_heads()
            if not heads:
                return None
            if len(heads) == 1:
                return heads[0]
            return ",".join(sorted(heads))
    finally:
        engine.dispose()


def database_migration_heads(
    database_url: str | None = None,
) -> tuple[str, ...]:
    url = database_url or settings.database_url
    engine = create_engine(url)
    try:
        with engine.connect() as connection:
            return tuple(
                sorted(MigrationContext.configure(connection).get_current_heads())
            )
    finally:
        engine.dispose()


def configured_migration_heads(
    database_url: str | None = None,
    *,
    enabled_modules: tuple[str, ...] | list[str] | None = None,
    manifest_factories: tuple[ManifestFactory, ...] = (),
    migration_track: str | None = None,
) -> tuple[str, ...]:
    config = alembic_config(
        database_url=database_url,
        enabled_modules=enabled_modules,
        manifest_factories=manifest_factories,
        migration_track=migration_track,
    )
    return tuple(sorted(ScriptDirectory.from_config(config).get_heads()))


def database_is_at_configured_heads(
    database_url: str | None = None,
    *,
    enabled_modules: tuple[str, ...] | list[str] | None = None,
    manifest_factories: tuple[ManifestFactory, ...] = (),
    migration_track: str | None = None,
) -> bool:
    return database_migration_heads(database_url) == configured_migration_heads(
        database_url,
        enabled_modules=enabled_modules,
        manifest_factories=manifest_factories,
        migration_track=migration_track,
    )


def _has_columns(inspector, table_name: str, required: set[str]) -> bool:
    try:
        actual = {column["name"] for column in inspector.get_columns(table_name)}
    except Exception:
        return False
    return required.issubset(actual)


def _has_indexes(inspector, table_name: str, required: set[str]) -> bool:
    try:
        actual = {index["name"] for index in inspector.get_indexes(table_name)}
    except Exception:
        return False
    return required.issubset(actual)


def _has_create_all_schema_through_hierarchical_settings(inspector, tables: set[str]) -> bool:
    if not _CREATE_ALL_THROUGH_HIERARCHICAL_TABLES.issubset(tables):
        return False
    if not any(
        table_name in tables and _has_columns(inspector, table_name, _SCOPE_TABLE_COLUMNS)
        for table_name in (CORE_SCOPE_TABLE, LEGACY_SCOPE_TABLE)
    ):
        return False
    for table_name, required_columns in _CREATE_ALL_THROUGH_HIERARCHICAL_COLUMNS.items():
        if table_name not in tables or not _has_columns(inspector, table_name, required_columns):
            return False
    for table_name, required_indexes in _CREATE_ALL_THROUGH_HIERARCHICAL_INDEXES.items():
        if table_name not in tables or not _has_indexes(inspector, table_name, required_indexes):
            return False
    return True


def _backfill_user_lock_state_for_create_all_schema(database_url: str) -> None:
    engine = create_engine(database_url)
    try:
        with engine.begin() as connection:
            connection.execute(text("""
                UPDATE campaign_versions
                SET user_lock_state = 'permanent',
                    user_locked_at = published_at,
                    user_locked_by_user_id = NULL
                WHERE published_at IS NOT NULL
                  AND user_lock_state IS NULL
            """))
    finally:
        engine.dispose()


def _row_count(connection, table_name: str) -> int:
    quoted = _quoted_table_name(connection, table_name)
    statement = text(f"SELECT COUNT(*) FROM {quoted}")  # noqa: S608  # nosec B608  # nosemgrep: python.sqlalchemy.security.audit.avoid-sqlalchemy-text.avoid-sqlalchemy-text
    return int(connection.execute(statement).scalar_one())


def _drop_table(connection, table_name: str) -> None:
    quoted = _quoted_table_name(connection, table_name)
    statement = text(f"DROP TABLE {quoted}")  # nosec B608  # nosemgrep: python.sqlalchemy.security.audit.avoid-sqlalchemy-text.avoid-sqlalchemy-text
    connection.execute(statement)


def _rename_table(connection, old_name: str, new_name: str) -> None:
    quoted_old = _quoted_table_name(connection, old_name)
    quoted_new = _quoted_table_name(connection, new_name)
    statement = text(f"ALTER TABLE {quoted_old} RENAME TO {quoted_new}")  # nosec B608  # nosemgrep: python.sqlalchemy.security.audit.avoid-sqlalchemy-text.avoid-sqlalchemy-text
    connection.execute(statement)


def _quoted_table_name(connection, table_name: str) -> str:
    if not _SQL_IDENTIFIER_RE.fullmatch(table_name):
        raise ValueError(f"Unsafe table identifier: {table_name!r}")
    return connection.dialect.identifier_preparer.quote(table_name)


def _reconcile_scope_table_names(connection, tables: set[str]) -> bool:
    if LEGACY_SCOPE_TABLE not in tables:
        return False

    changed = False
    if CORE_SCOPE_TABLE in tables:
        if _row_count(connection, CORE_SCOPE_TABLE) == 0:
            _drop_table(connection, CORE_SCOPE_TABLE)
            tables.remove(CORE_SCOPE_TABLE)
            changed = True
        elif _row_count(connection, LEGACY_SCOPE_TABLE) == 0:
            _drop_table(connection, LEGACY_SCOPE_TABLE)
            tables.remove(LEGACY_SCOPE_TABLE)
            changed = True
        else:
            raise RuntimeError(f"Cannot reconcile non-empty {LEGACY_SCOPE_TABLE} over non-empty {CORE_SCOPE_TABLE}")

    if LEGACY_SCOPE_TABLE in tables and CORE_SCOPE_TABLE not in tables:
        _rename_table(connection, LEGACY_SCOPE_TABLE, CORE_SCOPE_TABLE)
        tables.remove(LEGACY_SCOPE_TABLE)
        tables.add(CORE_SCOPE_TABLE)
        changed = True
    return changed


def reconcile_scope_table_retirement_drift(database_url: str | None = None) -> bool:
    """Repair databases that are stamped after the scope-table rename.

    The real Alembic migration performs the same rename. This helper covers
    development databases that were manually stamped or partially created via
    create_all(), where the version marker can be newer than the table name.
    It runs after Alembic upgrades so older migrations can still use the
    historical table name while they execute.
    """

    url = database_url or settings.database_url
    changed = False
    engine = create_engine(url)
    try:
        with engine.begin() as connection:
            tables = set(inspect(connection).get_table_names())
            changed = _reconcile_scope_table_names(connection, tables)
    finally:
        engine.dispose()
    return changed


def reconcile_namespace_table_drift(database_url: str | None = None) -> bool:
    """Repair dev databases stamped past the namespace-table migration.

    During the repository split some development databases were stamped at the
    newer Alembic heads while still carrying the old platform table names. The
    real migration only renames tables, so replay that idempotent operation
    before startup bootstrap code touches the ORM.
    """

    url = database_url or settings.database_url
    changed = False
    engine = create_engine(url)
    try:
        with engine.begin() as connection:
            schema = inspect(connection)
            tables = set(schema.get_table_names())
            if not any(old_name in tables and new_name not in tables for old_name, new_name in _NAMESPACE_TABLE_RENAMES):
                return False

            old_tables_to_drop: set[str] = set()
            new_tables_to_drop: set[str] = set()
            for old_name, new_name in _NAMESPACE_TABLE_RENAMES:
                if old_name not in tables or new_name not in tables:
                    continue
                if _row_count(connection, old_name) == 0:
                    old_tables_to_drop.add(old_name)
                elif _row_count(connection, new_name) == 0:
                    new_tables_to_drop.add(new_name)
                else:
                    raise RuntimeError(f"Cannot reconcile non-empty {old_name} over non-empty {new_name}")

            for old_name, _new_name in reversed(_NAMESPACE_TABLE_RENAMES):
                if old_name in old_tables_to_drop:
                    _drop_table(connection, old_name)
                    tables.remove(old_name)
                    changed = True
            for _old_name, new_name in reversed(_NAMESPACE_TABLE_RENAMES):
                if new_name in new_tables_to_drop:
                    _drop_table(connection, new_name)
                    tables.remove(new_name)
                    changed = True

            for old_name, new_name in _NAMESPACE_TABLE_RENAMES:
                if old_name not in tables or new_name in tables:
                    continue
                _rename_table(connection, old_name, new_name)
                tables.remove(old_name)
                tables.add(new_name)
                changed = True
    finally:
        engine.dispose()
    return changed


def reconcile_change_sequence_retention_floor_drift(database_url: str | None = None) -> bool:
    """Repair databases stamped after the change-sequence migration changed.

    Early development databases may have applied the change-sequence revision
    before the retention-floor table was added to that migration. The main
    change-sequence table is enough for full snapshots, but incremental delta
    requests need the retention floor to decide whether a watermark is stale.
    """

    url = database_url or settings.database_url
    changed = False
    engine = create_engine(url)
    try:
        with engine.begin() as connection:
            schema = inspect(connection)
            tables = set(schema.get_table_names())
            if ChangeSequenceEntry.__tablename__ not in tables:
                return False
            if ChangeSequenceRetentionFloor.__tablename__ not in tables:
                ChangeSequenceRetentionFloor.__table__.create(bind=connection, checkfirst=True)
                changed = True
            else:
                indexes = {
                    index["name"]
                    for index in schema.get_indexes(ChangeSequenceRetentionFloor.__tablename__)
                }
                for index in ChangeSequenceRetentionFloor.__table__.indexes:
                    if index.name not in indexes:
                        index.create(bind=connection)
                        changed = True
    finally:
        engine.dispose()
    return changed


def reconcile_covered_alembic_dependency_heads(
    database_url: str | None = None,
    *,
    config: Config | None = None,
) -> tuple[str, ...]:
    """Remove stale parent/dependency rows from ``alembic_version``.

    Alembic's version table represents current graph heads, not every owner
    head that contributed to the schema. After the v0.1.7 baseline squash,
    development databases may still carry dependency parents such as the core
    baseline beside child module heads. That shape is semantically already
    upgraded, but Alembic rejects it as an overlapping current revision set.
    """

    url = database_url or settings.database_url
    active_config = config or alembic_config(database_url=url)
    script = ScriptDirectory.from_config(active_config)
    engine = create_engine(url)
    removed: set[str] = set()
    try:
        with engine.begin() as connection:
            tables = set(inspect(connection).get_table_names())
            if "alembic_version" not in tables:
                return ()
            current = {
                str(row[0])
                for row in connection.execute(text("SELECT version_num FROM alembic_version")).all()
                if row[0]
            }
            if len(current) < 2:
                return ()
            covered: set[str] = set()
            for revision_id in current:
                try:
                    revision = script.get_revision(revision_id)
                except Exception as exc:
                    logger.debug("Skipping Alembic revision %s while pruning dependency heads: %s", revision_id, exc, exc_info=True)
                    continue
                ancestors = {
                    item.revision
                    for item in script.revision_map._get_ancestor_nodes(  # noqa: SLF001 - Alembic has no public dependency-ancestor API.
                        [revision],
                        include_dependencies=True,
                    )
                }
                covered.update(ancestors - {revision_id})
            removed = current.intersection(covered)
            for revision_id in sorted(removed):
                connection.execute(
                    text("DELETE FROM alembic_version WHERE version_num = :revision"),
                    {"revision": revision_id},
                )
    finally:
        engine.dispose()
    return tuple(sorted(removed))


def reconcile_legacy_create_all_schema(
    database_url: str | None = None,
    *,
    migration_track: str | None = None,
) -> str | None:
    """Repair the known Alembic/create_all drift without modifying application data.

    Returns the revision stamped during reconciliation, or ``None`` when no
    repair was necessary. A partial/unknown schema is intentionally left alone
    so Alembic can fail visibly instead of guessing.
    """

    url = database_url or settings.database_url
    state = _legacy_create_all_schema_state(url)
    target = _legacy_create_all_reconciliation_target(state)
    if target is None:
        return None
    if target == REVISION_HIERARCHICAL_SETTINGS:
        _backfill_user_lock_state_for_create_all_schema(url)

    command.stamp(alembic_config(database_url=url, migration_track=migration_track), target)
    return target


def _legacy_create_all_schema_state(database_url: str) -> _LegacyCreateAllSchemaState:
    engine = create_engine(database_url)
    try:
        with engine.connect() as connection:
            heads = MigrationContext.configure(connection).get_current_heads()
            schema = inspect(connection)
            tables = set(schema.get_table_names())
            return _legacy_create_all_schema_state_from_inspection(heads, schema, tables)
    finally:
        engine.dispose()


def _legacy_create_all_schema_state_from_inspection(
    heads: tuple[str, ...],
    schema: object,
    tables: set[str],
) -> _LegacyCreateAllSchemaState:
    has_file_storage = _FILE_STORAGE_TABLES.issubset(tables) and all(
        _has_columns(schema, table, _FILE_STORAGE_COLUMNS[table])
        for table in _FILE_STORAGE_TABLES
    )
    has_file_folders = _FILE_FOLDER_TABLES.issubset(tables) and _has_columns(
        schema,
        "file_folders",
        _FILE_STORAGE_COLUMNS["file_folders"],
    )
    return _LegacyCreateAllSchemaState(
        current=heads[0] if len(heads) == 1 else None,
        has_no_revision=len(heads) == 0,
        has_file_storage=has_file_storage,
        has_file_folders=has_file_folders,
        has_create_all_hierarchical_schema=_has_create_all_schema_through_hierarchical_settings(schema, tables),
    )


def _legacy_create_all_reconciliation_target(state: _LegacyCreateAllSchemaState) -> str | None:
    if state.current == REVISION_AUTH_RBAC and state.has_file_storage and state.has_file_folders:
        return REVISION_FILE_FOLDERS
    if state.current == REVISION_AUTH_RBAC and state.has_file_storage:
        return REVISION_FILE_STORAGE
    if state.current == REVISION_FILE_STORAGE and state.has_file_folders:
        return REVISION_FILE_FOLDERS
    if state.current == REVISION_FILE_FOLDERS and state.has_create_all_hierarchical_schema:
        return REVISION_HIERARCHICAL_SETTINGS
    if state.has_no_revision and state.has_create_all_hierarchical_schema:
        return REVISION_HIERARCHICAL_SETTINGS
    if state.has_no_revision and state.has_file_storage and state.has_file_folders:
        return REVISION_FILE_FOLDERS
    return None


def migrate_database(
    *,
    database_url: str | None = None,
    reconcile_legacy_schema: bool = True,
    enabled_modules: tuple[str, ...] | list[str] | None = None,
    migration_module_order: tuple[str, ...] | list[str] | None = None,
    manifest_factories: tuple[ManifestFactory, ...] = (),
    migration_track: str | None = None,
) -> MigrationResult:
    url = database_url or settings.database_url
    active_migration_track = _normalize_migration_track(migration_track)
    if reconcile_legacy_schema:
        reconcile_namespace_table_drift(url)
        reconcile_change_sequence_retention_floor_drift(url)
    previous = database_revision(url)
    reconciled = (
        reconcile_legacy_create_all_schema(url, migration_track=active_migration_track)
        if reconcile_legacy_schema
        else None
    )
    config = alembic_config(
        database_url=url,
        enabled_modules=enabled_modules,
        manifest_factories=manifest_factories,
        migration_track=active_migration_track,
    )
    reconcile_covered_alembic_dependency_heads(url, config=config)
    if migration_module_order:
        plan = registered_module_migration_plan(
            url,
            enabled_modules=enabled_modules,
            manifest_factories=manifest_factories,
            migration_track=active_migration_track,
        )
        _upgrade_ordered_module_heads(config, plan, tuple(migration_module_order))
    command.upgrade(config, "heads")
    if reconcile_legacy_schema:
        reconcile_scope_table_retirement_drift(url)
    current = database_revision(url)
    return MigrationResult(
        previous_revision=previous,
        reconciled_revision=reconciled,
        current_revision=current,
    )


def _upgrade_ordered_module_heads(
    config: Config,
    plan: MigrationMetadataPlan,
    module_order: tuple[str, ...],
) -> None:
    module_heads = _module_heads_by_id(config, plan)
    upgraded: set[str] = set()
    for module_id in dict.fromkeys(module_order):
        for head in module_heads.get(module_id, ()):
            if head in upgraded:
                continue
            command.upgrade(config, head)
            upgraded.add(head)


def _module_heads_by_id(config: Config, plan: MigrationMetadataPlan) -> dict[str, tuple[str, ...]]:
    script = ScriptDirectory.from_config(config)
    heads = script.get_heads()
    module_locations = {
        item.module_id: Path(item.script_location).resolve()
        for item in plan.modules
        if item.script_location
    }
    grouped: dict[str, list[str]] = {module_id: [] for module_id in module_locations}
    for head in heads:
        revision = script.get_revision(head)
        path_text = getattr(revision, "path", None)
        if revision is None or not path_text:
            continue
        revision_path = Path(path_text).resolve()
        for module_id, location in module_locations.items():
            try:
                revision_path.relative_to(location)
            except ValueError:
                continue
            grouped[module_id].append(head)
            break
    return {module_id: tuple(sorted(values)) for module_id, values in grouped.items() if values}
