"""drop mail credential override policy fields

Revision ID: 1b2c3d4e5f70
Revises: 0a1b2c3d4e6f
Create Date: 2026-06-25 18:30:00.000000
"""
from __future__ import annotations

import json
from typing import Any

from alembic import op
import sqlalchemy as sa

revision = "1b2c3d4e5f70"
down_revision = "0a1b2c3d4e6f"
branch_labels = None
depends_on = None

_MAIL_POLICY_KEY = "mail_profile_policy"
_CREDENTIAL_KEYS = ("smtp_credentials", "imap_credentials")
_DEPRECATED_LIMIT_KEYS = ("smtp_credentials.allow_override", "imap_credentials.allow_override")


def _json_table(table_name: str, column_name: str) -> sa.Table:
    return sa.table(
        table_name,
        sa.column("id", sa.String(length=36)),
        sa.column(column_name, sa.JSON()),
    )


def _as_dict(value: Any) -> dict[str, Any] | None:
    if isinstance(value, dict):
        return dict(value)
    if isinstance(value, str):
        try:
            parsed = json.loads(value)
        except json.JSONDecodeError:
            return None
        return dict(parsed) if isinstance(parsed, dict) else None
    return None


def _scrub_policy(value: Any) -> tuple[dict[str, Any] | None, bool]:
    policy = _as_dict(value)
    if policy is None:
        return None, False
    changed = False

    for key in _CREDENTIAL_KEYS:
        credential = _as_dict(policy.get(key))
        if credential is not None and "allow_override" in credential:
            credential.pop("allow_override", None)
            policy[key] = credential
            changed = True

    limits = _as_dict(policy.get("allow_lower_level_limits"))
    if limits is not None:
        for key in _DEPRECATED_LIMIT_KEYS:
            if key in limits:
                limits.pop(key, None)
                changed = True
        if changed:
            policy["allow_lower_level_limits"] = limits

    return policy, changed


def _scrub_settings_table(table_name: str) -> None:
    bind = op.get_bind()
    table = _json_table(table_name, "settings")
    rows = bind.execute(sa.select(table.c.id, table.c.settings)).mappings().all()
    for row in rows:
        settings = _as_dict(row["settings"])
        if settings is None:
            continue
        policy, changed = _scrub_policy(settings.get(_MAIL_POLICY_KEY))
        if changed and policy is not None:
            settings[_MAIL_POLICY_KEY] = policy
            bind.execute(sa.update(table).where(table.c.id == row["id"]).values(settings=settings))


def _scrub_policy_column(table_name: str) -> None:
    bind = op.get_bind()
    table = _json_table(table_name, "mail_profile_policy")
    rows = bind.execute(sa.select(table.c.id, table.c.mail_profile_policy)).mappings().all()
    for row in rows:
        policy, changed = _scrub_policy(row["mail_profile_policy"])
        if changed and policy is not None:
            bind.execute(sa.update(table).where(table.c.id == row["id"]).values(mail_profile_policy=policy))


def upgrade() -> None:
    inspector = sa.inspect(op.get_bind())
    tables = set(inspector.get_table_names())
    for table_name in ("core_system_settings", "tenancy_tenants"):
        if table_name in tables and "settings" in {column["name"] for column in inspector.get_columns(table_name)}:
            _scrub_settings_table(table_name)
    for table_name in ("access_users", "access_groups", "campaigns"):
        if table_name in tables and "mail_profile_policy" in {column["name"] for column in inspector.get_columns(table_name)}:
            _scrub_policy_column(table_name)


def downgrade() -> None:
    # The removed fields are redundant with the surviving lower-level limit
    # switches, so they cannot be reconstructed safely.
    pass
