leocrm/alembic/versions/0039_contact_normalize.py

"""Normalize contact model: fix surfix typo, Float→Numeric(5,2) discounts, JSON→JSONB, unique constraints.

Revision ID: 0039_contact_normalize
Revises: 0038_dms_content_hash
Create Date: 2026-07-25

Changes:
1. Rename column surfix → suffix on contacts table.
2. Convert discount_* columns from Float to Numeric(5,2) with CHECK constraints (0-100).
3. Convert custom columns from JSON to JSONB on contacts and contactpersons.
4. Add partial unique constraints: (tenant_id, code) and (tenant_id, accounting_code) where NOT NULL.
"""

from __future__ import annotations

from typing import Union

import sqlalchemy as sa
from alembic import op

# revision identifiers
revision: str = "0039_contact_normalize"
down_revision: Union[str, None] = "0038_dms_content_hash"
branch_labels: Union[str, None] = None
depends_on: Union[str, None] = None


DISCOUNT_COLUMNS = [
    "discount_crew",
    "discount_transport",
    "discount_rental",
    "discount_sale",
    "discount_subrent",
    "discount_total",
]


def _column_exists(table: str, column: str) -> str:
    """Return SQL that checks if a column exists on a table."""
    return (
        f"SELECT 1 FROM information_schema.columns "
        f"WHERE table_name = '{table}' AND column_name = '{column}'"
    )


def _constraint_exists(table: str, constraint: str) -> str:
    """Return SQL that checks if a constraint exists on a table."""
    return (
        f"SELECT 1 FROM information_schema.table_constraints "
        f"WHERE table_name = '{table}' AND constraint_name = '{constraint}'"
    )


def upgrade() -> None:
    conn = op.get_bind()

    # ── 0. Add status column to contacts (for state machine) ──
    status_col = conn.execute(sa.text(_column_exists("contacts", "status"))).fetchone()
    if not status_col:
        op.add_column("contacts", sa.Column("status", sa.String(20), nullable=False, server_default="lead"))

    # ── 1a. Rename surfix → suffix ──
    result = conn.execute(sa.text(_column_exists("contacts", "surfix"))).fetchone()
    if result:
        op.alter_column("contacts", "surfix", new_column_name="suffix")

    # ── 1b. Convert discount_* from Float to Numeric(5,2) with CHECK ──
    for col in DISCOUNT_COLUMNS:
        conn.execute(
            sa.text(
                f"ALTER TABLE contacts ALTER COLUMN {col} "
                f"TYPE NUMERIC(5,2) USING {col}::numeric(5,2)"
            )
        )
        # Add CHECK constraint if not exists
        ck_name = f"ck_contacts_{col}_range"
        ck_exists = conn.execute(
            sa.text(_constraint_exists("contacts", ck_name))
        ).fetchone()
        if not ck_exists:
            conn.execute(
                sa.text(
                    f"ALTER TABLE contacts ADD CONSTRAINT {ck_name} "
                    f"CHECK ({col} BETWEEN 0 AND 100)"
                )
            )

    # ── 1c. JSON → JSONB for contacts.custom ──
    result = conn.execute(
        sa.text(
            "SELECT data_type FROM information_schema.columns "
            "WHERE table_name = 'contacts' AND column_name = 'custom'"
        )
    ).fetchone()
    if result and result[0] == "json":
        conn.execute(
            sa.text(
                "ALTER TABLE contacts ALTER COLUMN custom "
                "TYPE JSONB USING custom::jsonb"
            )
        )

    # ── 1d. JSON → JSONB for contactpersons.custom ──
    result = conn.execute(
        sa.text(
            "SELECT data_type FROM information_schema.columns "
            "WHERE table_name = 'contactpersons' AND column_name = 'custom'"
        )
    ).fetchone()
    if result and result[0] == "json":
        conn.execute(
            sa.text(
                "ALTER TABLE contactpersons ALTER COLUMN custom "
                "TYPE JSONB USING custom::jsonb"
            )
        )

    # ── 1e. Partial unique constraints ──
    # (tenant_id, code) where code IS NOT NULL
    uq_code_exists = conn.execute(
        sa.text(_constraint_exists("contacts", "uq_contacts_tenant_code"))
    ).fetchone()
    if not uq_code_exists:
        conn.execute(
            sa.text(
                "CREATE UNIQUE INDEX uq_contacts_tenant_code "
                "ON contacts (tenant_id, code) WHERE code IS NOT NULL"
            )
        )

    # (tenant_id, accounting_code) where accounting_code IS NOT NULL
    uq_acct_exists = conn.execute(
        sa.text(_constraint_exists("contacts", "uq_contacts_tenant_accounting_code"))
    ).fetchone()
    if not uq_acct_exists:
        conn.execute(
            sa.text(
                "CREATE UNIQUE INDEX uq_contacts_tenant_accounting_code "
                "ON contacts (tenant_id, accounting_code) WHERE accounting_code IS NOT NULL"
            )
        )


def downgrade() -> None:
    conn = op.get_bind()

    # Drop unique indexes
    conn.execute(sa.text("DROP INDEX IF EXISTS uq_contacts_tenant_accounting_code"))
    conn.execute(sa.text("DROP INDEX IF EXISTS uq_contacts_tenant_code"))

    # JSONB → JSON
    conn.execute(
        sa.text(
            "ALTER TABLE contactpersons ALTER COLUMN custom "
            "TYPE JSON USING custom::json"
        )
    )
    conn.execute(
        sa.text(
            "ALTER TABLE contacts ALTER COLUMN custom TYPE JSON USING custom::json"
        )
    )

    # Drop CHECK constraints and revert Numeric → Float
    for col in DISCOUNT_COLUMNS:
        ck_name = f"ck_contacts_{col}_range"
        conn.execute(sa.text(f"ALTER TABLE contacts DROP CONSTRAINT IF EXISTS {ck_name}"))
        conn.execute(
            sa.text(
                f"ALTER TABLE contacts ALTER COLUMN {col} "
                f"TYPE FLOAT USING {col}::float"
            )
        )

    # Rename suffix → surfix
    result = conn.execute(sa.text(_column_exists("contacts", "suffix"))).fetchone()
    if result:
        op.alter_column("contacts", "suffix", new_column_name="surfix")