"""add multi telephony config tables

Revision ID: a2355fc6bdc1
Revises: 4d8e9b2a3c5f
Create Date: 2026-04-26 15:07:07.644855

"""

import json
import re
from dataclasses import dataclass
from typing import Optional, Sequence, Union

import sqlalchemy as sa
from alembic import op

# revision identifiers, used by Alembic.
revision: str = "a2355fc6bdc1"
down_revision: Union[str, None] = "4d8e9b2a3c5f"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None


# Credential keys that are NOT provider credentials and must be stripped before
# the legacy TELEPHONY_CONFIGURATION JSON is copied into the new credentials column.
_NON_CREDENTIAL_KEYS = {"provider", "from_numbers"}


# Self-contained address normalizer for the legacy ``from_numbers`` backfill.
# Migrations are snapshots — we deliberately avoid importing the live
# ``api.utils.telephony_address`` here so this migration keeps working even if
# that module later moves or its signature changes. The legacy data only ever
# reached this column without country hints, so the country-hint branch of the
# real normalizer is intentionally omitted.
_PSTN_DIGITS_RE = re.compile(r"^\d{8,15}$")
_PSTN_STRIP_RE = re.compile(r"[\s\-\(\)]")
_SIP_URI_RE = re.compile(
    r"^(?P<scheme>sips?):(?:(?P<user>[^@;?]+)@)?(?P<host>[^:;?]+)"
    r"(?::(?P<port>\d+))?(?P<rest>[;?].*)?$",
    re.IGNORECASE,
)


@dataclass(frozen=True)
class _NormalizedAddress:
    canonical: str
    address_type: str
    country_code: Optional[str] = None


def normalize_telephony_address(raw: str) -> _NormalizedAddress:
    if raw is None:
        raise ValueError("address must not be None")
    raw = raw.strip()
    if not raw:
        raise ValueError("address must not be empty")

    if raw.lower().startswith(("sip:", "sips:")):
        m = _SIP_URI_RE.match(raw)
        if not m:
            return _NormalizedAddress(canonical=raw.lower(), address_type="sip_uri")
        scheme = m.group("scheme").lower()
        user = m.group("user")
        host = m.group("host").lower()
        port = m.group("port")
        rest = m.group("rest") or ""
        if (scheme == "sip" and port == "5060") or (
            scheme == "sips" and port == "5061"
        ):
            port = None
        canonical = f"{scheme}:"
        if user:
            canonical += f"{user}@"
        canonical += host
        if port:
            canonical += f":{port}"
        if rest:
            canonical += rest.lower()
        return _NormalizedAddress(canonical=canonical, address_type="sip_uri")

    digits = _PSTN_STRIP_RE.sub("", raw)
    if digits.startswith("+"):
        digits = digits[1:]
    if _PSTN_DIGITS_RE.fullmatch(digits):
        return _NormalizedAddress(canonical=f"+{digits}", address_type="pstn")

    return _NormalizedAddress(canonical=raw.lower(), address_type="sip_extension")


def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_table(
        "telephony_configurations",
        sa.Column("id", sa.Integer(), nullable=False),
        sa.Column("organization_id", sa.Integer(), nullable=False),
        sa.Column("name", sa.String(length=64), nullable=False),
        sa.Column("provider", sa.String(length=32), nullable=False),
        sa.Column("credentials", sa.JSON(), nullable=False),
        sa.Column(
            "is_default_outbound",
            sa.Boolean(),
            server_default=sa.text("false"),
            nullable=False,
        ),
        sa.Column("created_at", sa.DateTime(timezone=True), nullable=True),
        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=True),
        sa.ForeignKeyConstraint(
            ["organization_id"], ["organizations.id"], ondelete="CASCADE"
        ),
        sa.PrimaryKeyConstraint("id"),
        sa.UniqueConstraint(
            "organization_id", "name", name="uq_telephony_configurations_org_name"
        ),
    )
    op.create_index(
        "ix_telephony_configurations_org",
        "telephony_configurations",
        ["organization_id"],
        unique=False,
    )
    op.create_index(
        "uq_telephony_configurations_default",
        "telephony_configurations",
        ["organization_id"],
        unique=True,
        postgresql_where=sa.text("is_default_outbound = true"),
    )
    op.create_table(
        "telephony_phone_numbers",
        sa.Column("id", sa.Integer(), nullable=False),
        sa.Column("organization_id", sa.Integer(), nullable=False),
        sa.Column("telephony_configuration_id", sa.Integer(), nullable=False),
        sa.Column("address", sa.String(length=255), nullable=False),
        sa.Column("address_normalized", sa.String(length=255), nullable=False),
        sa.Column("address_type", sa.String(length=16), nullable=False),
        sa.Column("country_code", sa.String(length=2), nullable=True),
        sa.Column("label", sa.String(length=64), nullable=True),
        sa.Column("inbound_workflow_id", sa.Integer(), nullable=True),
        sa.Column(
            "is_active", sa.Boolean(), server_default=sa.text("true"), nullable=False
        ),
        sa.Column(
            "is_default_caller_id",
            sa.Boolean(),
            server_default=sa.text("false"),
            nullable=False,
        ),
        sa.Column(
            "extra_metadata",
            sa.JSON(),
            server_default=sa.text("'{}'::json"),
            nullable=False,
        ),
        sa.Column("created_at", sa.DateTime(timezone=True), nullable=True),
        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=True),
        sa.ForeignKeyConstraint(
            ["inbound_workflow_id"], ["workflows.id"], ondelete="SET NULL"
        ),
        sa.ForeignKeyConstraint(
            ["organization_id"], ["organizations.id"], ondelete="CASCADE"
        ),
        sa.ForeignKeyConstraint(
            ["telephony_configuration_id"],
            ["telephony_configurations.id"],
            ondelete="CASCADE",
        ),
        sa.PrimaryKeyConstraint("id"),
        sa.UniqueConstraint(
            "organization_id", "address_normalized", name="uq_phone_numbers_org_address"
        ),
    )
    op.create_index(
        "ix_phone_numbers_config",
        "telephony_phone_numbers",
        ["telephony_configuration_id"],
        unique=False,
    )
    op.create_index(
        "ix_phone_numbers_inbound_lookup",
        "telephony_phone_numbers",
        ["address_normalized", "organization_id"],
        unique=False,
        postgresql_where=sa.text("is_active = true"),
    )
    op.create_index(
        "ix_phone_numbers_workflow",
        "telephony_phone_numbers",
        ["inbound_workflow_id"],
        unique=False,
        postgresql_where=sa.text("inbound_workflow_id IS NOT NULL"),
    )
    op.create_index(
        "uq_phone_numbers_default_caller",
        "telephony_phone_numbers",
        ["telephony_configuration_id"],
        unique=True,
        postgresql_where=sa.text("is_default_caller_id = true"),
    )
    op.add_column(
        "campaigns",
        sa.Column("telephony_configuration_id", sa.Integer(), nullable=True),
    )
    op.create_index(
        "ix_campaigns_telephony_config",
        "campaigns",
        ["telephony_configuration_id"],
        unique=False,
        postgresql_where=sa.text("telephony_configuration_id IS NOT NULL"),
    )
    op.create_foreign_key(
        "fk_campaigns_telephony_configuration_id",
        "campaigns",
        "telephony_configurations",
        ["telephony_configuration_id"],
        ["id"],
    )
    # ### end Alembic commands ###

    _backfill_from_legacy_telephony_configuration()


def _backfill_from_legacy_telephony_configuration() -> None:
    """Copy legacy ``OrganizationConfigurationModel(TELEPHONY_CONFIGURATION)`` rows
    into the new tables and back-link campaigns to each org's default config.

    For every legacy row we create:
      - one ``telephony_configurations`` row (name='Default', is_default_outbound=true),
      - one ``telephony_phone_numbers`` row per entry in ``value['from_numbers']``.

    Then ``campaigns.telephony_configuration_id`` is set to that org's default config
    so the dispatcher's lookup keeps working without a code-side fallback.
    """
    bind = op.get_bind()

    legacy_rows = bind.execute(
        sa.text(
            "SELECT organization_id, value FROM organization_configurations "
            "WHERE key = 'TELEPHONY_CONFIGURATION'"
        )
    ).fetchall()

    for org_id, raw_value in legacy_rows:
        value = (
            raw_value if isinstance(raw_value, dict) else json.loads(raw_value or "{}")
        )
        if not value:
            continue

        provider = value.get("provider")
        if not provider:
            continue

        credentials = {k: v for k, v in value.items() if k not in _NON_CREDENTIAL_KEYS}

        # Set created_at/updated_at explicitly: the SQLAlchemy model's
        # ``default=lambda: datetime.now(UTC)`` only fires on ORM-driven inserts,
        # not on raw SQL like this. Without these, both columns are NULL and the
        # ORM read path fails Pydantic validation.
        cfg_id = bind.execute(
            sa.text(
                "INSERT INTO telephony_configurations "
                "(organization_id, name, provider, credentials, is_default_outbound, "
                " created_at, updated_at) "
                "VALUES (:org_id, 'Default', :provider, CAST(:creds AS JSON), TRUE, "
                " NOW(), NOW()) "
                "RETURNING id"
            ),
            {
                "org_id": org_id,
                "provider": provider,
                "creds": json.dumps(credentials),
            },
        ).scalar_one()

        for raw_address in value.get("from_numbers") or []:
            if raw_address is None:
                continue
            try:
                normalized = normalize_telephony_address(str(raw_address))
            except ValueError:
                continue

            # Skip duplicates within the same org. The unique constraint would also
            # catch this; we pre-filter to avoid breaking the whole transaction on
            # legacy data with the same number listed twice.
            existing = bind.execute(
                sa.text(
                    "SELECT 1 FROM telephony_phone_numbers "
                    "WHERE organization_id = :org_id AND address_normalized = :addr"
                ),
                {"org_id": org_id, "addr": normalized.canonical},
            ).first()
            if existing:
                continue

            bind.execute(
                sa.text(
                    "INSERT INTO telephony_phone_numbers "
                    "(organization_id, telephony_configuration_id, address, "
                    " address_normalized, address_type, country_code, "
                    " created_at, updated_at) "
                    "VALUES (:org_id, :cfg_id, :addr, :norm, :type, :cc, "
                    " NOW(), NOW())"
                ),
                {
                    "org_id": org_id,
                    "cfg_id": cfg_id,
                    "addr": str(raw_address),
                    "norm": normalized.canonical,
                    "type": normalized.address_type,
                    "cc": normalized.country_code,
                },
            )

    # Backfill campaigns to point at their org's default config.
    bind.execute(
        sa.text(
            "UPDATE campaigns c "
            "SET telephony_configuration_id = tc.id "
            "FROM telephony_configurations tc "
            "WHERE tc.organization_id = c.organization_id "
            "  AND tc.is_default_outbound = TRUE "
            "  AND c.telephony_configuration_id IS NULL"
        )
    )

    _move_ari_inbound_workflow_to_phone_numbers()

    _validate_migrated_configurations()


def _move_ari_inbound_workflow_to_phone_numbers() -> None:
    """ARI's legacy single-config-per-org model stored the inbound workflow as
    ``credentials.inbound_workflow_id`` — one workflow for the whole connection
    regardless of extension. The multi-config schema puts inbound routing on
    ``telephony_phone_numbers`` (one workflow per extension), matching every
    other provider, so the ARI request/response models no longer declare the
    field. Copy it onto each linked phone number, then strip it from
    credentials so the data and the schema agree before validation runs.
    """
    # 1. Backfill telephony_phone_numbers.inbound_workflow_id from each ARI
    #    config's credentials.inbound_workflow_id. Only fill NULLs — the
    #    inserts above leave this column NULL for every newly-created row,
    #    but the guard keeps the step idempotent if rerun.
    op.execute(
        """
        UPDATE telephony_phone_numbers tpn
        SET inbound_workflow_id = ((tc.credentials::jsonb)->>'inbound_workflow_id')::integer,
            updated_at = NOW()
        FROM telephony_configurations tc
        WHERE tpn.telephony_configuration_id = tc.id
          AND tc.provider = 'ari'
          AND (tc.credentials::jsonb) ? 'inbound_workflow_id'
          AND ((tc.credentials::jsonb)->>'inbound_workflow_id') ~ '^[0-9]+$'
          AND tpn.inbound_workflow_id IS NULL
        """
    )

    # 2. Strip the legacy key from ARI configs' credentials so the schema and
    #    the data agree (the ARI provider's request/response models no longer
    #    declare inbound_workflow_id, which would otherwise trip the
    #    extra-fields check in _validate_migrated_configurations).
    op.execute(
        """
        UPDATE telephony_configurations
        SET credentials = ((credentials::jsonb) - 'inbound_workflow_id')::json,
            updated_at = NOW()
        WHERE provider = 'ari'
          AND (credentials::jsonb) ? 'inbound_workflow_id'
        """
    )


_PLACEHOLDER_VALUE = "CHANGE_ME"


def _validate_migrated_configurations() -> None:
    """Round-trip every migrated row through its provider's Pydantic
    request schema so legacy data that the live code can no longer parse
    fails the migration loudly instead of silently breaking at runtime.

    Migrations normally avoid importing live application modules to stay
    replay-safe, but the whole point of this step is to compare the
    just-written rows against the *current* schemas — so the import is
    intentional. If a provider has been removed or renamed since the
    legacy data was saved, that surfaces here as a missing-spec failure.

    Validation mirrors what the runtime sees: ``provider`` + the JSONB
    ``credentials`` columns + ``from_numbers`` joined from
    ``telephony_phone_numbers``. We additionally reject credential keys
    that aren't declared on the request model — Pydantic ignores extras
    by default, so a stray legacy field (e.g. a renamed credential) would
    otherwise slip through and only show up later as a confused operator
    wondering why a value they entered does nothing.

    Required string fields that are *missing* from legacy credentials
    (e.g. Plivo's ``application_id`` or Cloudonix's ``application_name``,
    which post-date some legacy rows) are filled with the
    ``"CHANGE_ME"`` placeholder and the row is rewritten in place. Without
    this, the next ORM read fails Pydantic validation and the operator's
    config form refuses to render — leaving them with no UI path to enter
    the missing value.
    """
    import importlib

    from pydantic import ValidationError

    from api.services.telephony import registry

    # Triggers each provider package's ``register()`` side effect.
    importlib.import_module("api.services.telephony.providers")

    bind = op.get_bind()

    cfg_rows = bind.execute(
        sa.text(
            "SELECT id, organization_id, name, provider, credentials "
            "FROM telephony_configurations"
        )
    ).fetchall()

    failures = []
    patched = []

    for cfg_id, org_id, cfg_name, provider, raw_credentials in cfg_rows:
        credentials = (
            raw_credentials
            if isinstance(raw_credentials, dict)
            else json.loads(raw_credentials or "{}")
        )

        spec = registry.get_optional(provider)
        if spec is None:
            failures.append(
                f"id={cfg_id} org={org_id} name={cfg_name!r}: provider "
                f"{provider!r} is not registered (registered: "
                f"{sorted(registry.names())})"
            )
            continue

        added_placeholders = []
        for field_name, field_info in spec.config_request_cls.model_fields.items():
            if field_name in {"provider", "from_numbers"}:
                continue
            if field_name in credentials:
                continue
            if not field_info.is_required():
                continue
            # Only string fields get the sentinel — a non-string required field
            # (e.g. an int port) would still fail validation with "CHANGE_ME"
            # and the placeholder would mislead the operator. Surface it as a
            # failure instead of silently writing a wrong-type value.
            if field_info.annotation is not str:
                continue
            credentials[field_name] = _PLACEHOLDER_VALUE
            added_placeholders.append(field_name)

        if added_placeholders:
            bind.execute(
                sa.text(
                    "UPDATE telephony_configurations "
                    "SET credentials = CAST(:creds AS JSON), updated_at = NOW() "
                    "WHERE id = :cfg_id"
                ),
                {"creds": json.dumps(credentials), "cfg_id": cfg_id},
            )
            patched.append(
                f"id={cfg_id} org={org_id} name={cfg_name!r} provider={provider!r}: "
                f"set placeholder {_PLACEHOLDER_VALUE!r} for missing required "
                f"field(s) {added_placeholders}"
            )

        from_numbers = [
            row[0]
            for row in bind.execute(
                sa.text(
                    "SELECT address FROM telephony_phone_numbers "
                    "WHERE telephony_configuration_id = :cfg_id"
                ),
                {"cfg_id": cfg_id},
            ).fetchall()
        ]

        # Explicit keys win over anything in credentials so a stray "provider"
        # or "from_numbers" left in the JSONB can't shadow the canonical values.
        payload = {**credentials, "provider": provider, "from_numbers": from_numbers}

        known_fields = set(spec.config_request_cls.model_fields)
        extras = sorted(set(credentials) - known_fields - {"provider", "from_numbers"})
        if extras:
            failures.append(
                f"id={cfg_id} org={org_id} name={cfg_name!r} provider={provider!r}: "
                f"credentials contain unknown field(s) {extras} not declared on "
                f"{spec.config_request_cls.__name__}"
            )
            continue

        try:
            spec.config_request_cls.model_validate(payload)
        except ValidationError as exc:
            failures.append(
                f"id={cfg_id} org={org_id} name={cfg_name!r} provider={provider!r} "
                f"failed {spec.config_request_cls.__name__} validation: {exc}"
            )

    if patched or failures:
        from loguru import logger

        if patched:
            logger.warning(
                "Migrated telephony configurations had missing required fields "
                f"filled with the {_PLACEHOLDER_VALUE!r} placeholder. Update "
                "these in the UI before relying on the affected providers:\n  - "
                + "\n  - ".join(patched)
            )

        if failures:
            logger.warning(
                "Migrated telephony configurations did not pass live Pydantic "
                "validation. The migration will continue, but these rows will "
                "fail at runtime until fixed in the new tables:\n  - "
                + "\n  - ".join(failures)
            )


def downgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_constraint(
        "fk_campaigns_telephony_configuration_id", "campaigns", type_="foreignkey"
    )
    op.drop_index(
        "ix_campaigns_telephony_config",
        table_name="campaigns",
        postgresql_where=sa.text("telephony_configuration_id IS NOT NULL"),
    )
    op.drop_column("campaigns", "telephony_configuration_id")
    op.drop_index(
        "uq_phone_numbers_default_caller",
        table_name="telephony_phone_numbers",
        postgresql_where=sa.text("is_default_caller_id = true"),
    )
    op.drop_index(
        "ix_phone_numbers_workflow",
        table_name="telephony_phone_numbers",
        postgresql_where=sa.text("inbound_workflow_id IS NOT NULL"),
    )
    op.drop_index(
        "ix_phone_numbers_inbound_lookup",
        table_name="telephony_phone_numbers",
        postgresql_where=sa.text("is_active = true"),
    )
    op.drop_index("ix_phone_numbers_config", table_name="telephony_phone_numbers")
    op.drop_table("telephony_phone_numbers")
    op.drop_index(
        "uq_telephony_configurations_default",
        table_name="telephony_configurations",
        postgresql_where=sa.text("is_default_outbound = true"),
    )
    op.drop_index(
        "ix_telephony_configurations_org", table_name="telephony_configurations"
    )
    op.drop_table("telephony_configurations")
    # ### end Alembic commands ###