leocrm/alembic/versions/0021_unified_contacts.py

"""Unified contacts model — company or person with inline addresses.

Revision ID: 0021
Revises: 0020
Create Date: 2026-07-19

SAFE MIGRATION: Old tables are renamed (not dropped), data is migrated
via INSERT ... SELECT, and old tables are preserved for rollback.
"""

from __future__ import annotations

import logging
from typing import Sequence, Union

from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects.postgresql import UUID, TSVECTOR, JSON

revision: str = "0021_unified_contacts"
down_revision: Union[str, None] = "0020"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None

logger = logging.getLogger("alembic.migration.0021")


def _table_exists(conn, table_name: str) -> bool:
    """Check whether *table_name* exists in the public schema."""
    result = conn.execute(
        sa.text(
            "SELECT 1 FROM information_schema.tables "
            "WHERE table_schema = 'public' AND table_name = :t"
        ),
        {"t": table_name},
    ).fetchone()
    return result is not None


def _column_exists(conn, table_name: str, column_name: str) -> bool:
    """Check whether *column_name* exists on *table_name*."""
    result = conn.execute(
        sa.text(
            "SELECT 1 FROM information_schema.columns "
            "WHERE table_schema = 'public' "
            "AND table_name = :t AND column_name = :c"
        ),
        {"t": table_name, "c": column_name},
    ).fetchone()
    return result is not None


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

    # ── 1. Rename old tables instead of dropping ──────────────────────
    #    Only rename if the table exists and the _old version doesn't.
    old_tables = ["company_contacts", "contacts", "companies"]
    renamed: list[str] = []

    for tbl in old_tables:
        old_name = f"{tbl}_old"
        if _table_exists(conn, tbl) and not _table_exists(conn, old_name):
            op.execute(f'ALTER TABLE "{tbl}" RENAME TO "{old_name}"')
            renamed.append(old_name)
            logger.info("Renamed %s → %s", tbl, old_name)
        elif _table_exists(conn, old_name):
            logger.info("%s already exists — skipping rename of %s", old_name, tbl)
        else:
            logger.info("Table %s does not exist — nothing to rename", tbl)

    # ── 2. Create new contacts table ──────────────────────────────────
    # Drop indexes that were carried over from the renamed old tables
    op.execute("DROP INDEX IF EXISTS ix_contacts_tenant_id")
    op.create_table(
        "contacts",
        sa.Column("id", UUID(as_uuid=True), primary_key=True, server_default=sa.text("gen_random_uuid()")),
        sa.Column("tenant_id", UUID(as_uuid=True), nullable=False, index=True),
        # Identity & Type
        sa.Column("type", sa.String(20), nullable=False, server_default="company"),
        sa.Column("displayname", sa.String(255), nullable=False, server_default=""),
        sa.Column("name", sa.String(255), nullable=True),
        sa.Column("firstname", sa.String(100), nullable=True),
        sa.Column("surname", sa.String(100), nullable=True),
        sa.Column("surfix", sa.String(50), nullable=True),
        sa.Column("ext_name_line", sa.String(255), nullable=True),
        sa.Column("gender", sa.String(20), nullable=True),
        # Customer / Accounting
        sa.Column("code", sa.String(100), nullable=True),
        sa.Column("accounting_code", sa.String(100), nullable=True),
        sa.Column("vendor_accounting_code", sa.String(100), nullable=True),
        # Mailing Address
        sa.Column("mailing_street", sa.String(255), nullable=True),
        sa.Column("mailing_number", sa.String(20), nullable=True),
        sa.Column("mailing_unit_number", sa.String(50), nullable=True),
        sa.Column("mailing_district", sa.String(100), nullable=True),
        sa.Column("mailing_extra_address_line", sa.String(255), nullable=True),
        sa.Column("mailing_postalcode", sa.String(20), nullable=True),
        sa.Column("mailing_city", sa.String(100), nullable=True),
        sa.Column("mailing_state", sa.String(100), nullable=True),
        sa.Column("mailing_country", sa.String(2), nullable=True),
        # Visit Address
        sa.Column("visit_street", sa.String(255), nullable=True),
        sa.Column("visit_number", sa.String(20), nullable=True),
        sa.Column("visit_unit_number", sa.String(50), nullable=True),
        sa.Column("visit_district", sa.String(100), nullable=True),
        sa.Column("visit_extra_address_line", sa.String(255), nullable=True),
        sa.Column("visit_postalcode", sa.String(20), nullable=True),
        sa.Column("visit_city", sa.String(100), nullable=True),
        sa.Column("visit_state", sa.String(100), nullable=True),
        # Invoice Address
        sa.Column("invoice_street", sa.String(255), nullable=True),
        sa.Column("invoice_number", sa.String(20), nullable=True),
        sa.Column("invoice_unit_number", sa.String(50), nullable=True),
        sa.Column("invoice_district", sa.String(100), nullable=True),
        sa.Column("invoice_extra_address_line", sa.String(255), nullable=True),
        sa.Column("invoice_postalcode", sa.String(20), nullable=True),
        sa.Column("invoice_city", sa.String(100), nullable=True),
        sa.Column("invoice_state", sa.String(100), nullable=True),
        sa.Column("invoice_country", sa.String(2), nullable=True),
        # General country
        sa.Column("country", sa.String(2), nullable=True),
        # Communication
        sa.Column("phone_1", sa.String(50), nullable=True),
        sa.Column("phone_2", sa.String(50), nullable=True),
        sa.Column("email_1", sa.String(255), nullable=True),
        sa.Column("email_2", sa.String(255), nullable=True),
        sa.Column("website", sa.String(500), nullable=True),
        # Financial & Tax
        sa.Column("vat_code", sa.String(50), nullable=True),
        sa.Column("fiscal_code", sa.String(50), nullable=True),
        sa.Column("commerce_code", sa.String(100), nullable=True),
        sa.Column("purchase_number", sa.String(100), nullable=True),
        sa.Column("bic", sa.String(50), nullable=True),
        sa.Column("bank_account", sa.String(50), nullable=True),
        # Discounts
        sa.Column("discount_crew", sa.Float, nullable=False, server_default="0"),
        sa.Column("discount_transport", sa.Float, nullable=False, server_default="0"),
        sa.Column("discount_rental", sa.Float, nullable=False, server_default="0"),
        sa.Column("discount_sale", sa.Float, nullable=False, server_default="0"),
        sa.Column("discount_subrent", sa.Float, nullable=False, server_default="0"),
        sa.Column("discount_total", sa.Float, nullable=False, server_default="0"),
        # Geo
        sa.Column("latitude", sa.Float, nullable=True),
        sa.Column("longitude", sa.Float, nullable=True),
        # Notes & Warnings
        sa.Column("projectnote", sa.Text, nullable=True),
        sa.Column("projectnote_title", sa.String(255), nullable=True),
        sa.Column("contact_warning", sa.Text, nullable=True),
        sa.Column("tags", sa.String(500), nullable=True),
        sa.Column("image", sa.Text, nullable=True),
        # Custom fields
        sa.Column("custom", JSON, nullable=True, server_default=sa.text("'{}'::json")),
        # FTS
        sa.Column("search_tsv", TSVECTOR, sa.Computed(
            "to_tsvector('german', coalesce(name, '') || ' ' || coalesce(displayname, '') || ' ' || coalesce(firstname, '') || ' ' || coalesce(surname, '') || ' ' || coalesce(email_1, '') || ' ' || coalesce(email_2, '') || ' ' || coalesce(code, '') || ' ' || coalesce(phone_1, '') || ' ' || coalesce(phone_2, '') || ' ' || coalesce(mailing_city, '') || ' ' || coalesce(mailing_postalcode, '') || ' ' || coalesce(tags, ''))",
            persisted=True,
        ), nullable=True),
        # Audit
        sa.Column("created_by", UUID(as_uuid=True), sa.ForeignKey("users.id", ondelete="SET NULL"), nullable=True),
        sa.Column("updated_by", UUID(as_uuid=True), sa.ForeignKey("users.id", ondelete="SET NULL"), nullable=True),
        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
        sa.Column("deleted_at", sa.DateTime(timezone=True), nullable=True),
    )
    op.execute("DROP INDEX IF EXISTS ix_contacts_tenant_deleted")
    op.execute("CREATE INDEX IF NOT EXISTS ix_contacts_tenant_deleted ON contacts (tenant_id, deleted_at)")
    op.execute('CREATE INDEX IF NOT EXISTS ix_contacts_tenant_type ON contacts (tenant_id, type)')
    op.execute("DROP INDEX IF EXISTS ix_contacts_tenant_name")
    op.execute("CREATE INDEX IF NOT EXISTS ix_contacts_tenant_name ON contacts (tenant_id, name)")
    op.execute('CREATE INDEX IF NOT EXISTS ix_contacts_tenant_displayname ON contacts (tenant_id, displayname)')
    op.execute("DROP INDEX IF EXISTS ix_contacts_email")
    op.execute("CREATE INDEX IF NOT EXISTS ix_contacts_email ON contacts (email_1)")
    op.execute('CREATE INDEX IF NOT EXISTS ix_contacts_code ON contacts (code)')
    op.execute('CREATE INDEX IF NOT EXISTS ix_contacts_search_vec ON contacts (search_tsv)')

    # ── 3. Create contactpersons table ────────────────────────────────
    op.create_table(
        "contactpersons",
        sa.Column("id", UUID(as_uuid=True), primary_key=True, server_default=sa.text("gen_random_uuid()")),
        sa.Column("tenant_id", UUID(as_uuid=True), nullable=False, index=True),
        sa.Column("contact_id", UUID(as_uuid=True), sa.ForeignKey("contacts.id", ondelete="CASCADE"), nullable=False),
        sa.Column("displayname", sa.String(255), nullable=False, server_default=""),
        sa.Column("firstname", sa.String(100), nullable=True),
        sa.Column("middle_name", sa.String(100), nullable=True),
        sa.Column("lastname", sa.String(100), nullable=True),
        sa.Column("function", sa.String(255), nullable=True),
        sa.Column("phone", sa.String(50), nullable=True),
        sa.Column("mobilephone", sa.String(50), nullable=True),
        sa.Column("email", sa.String(255), nullable=True),
        sa.Column("street", sa.String(255), nullable=True),
        sa.Column("number", sa.String(20), nullable=True),
        sa.Column("postalcode", sa.String(20), nullable=True),
        sa.Column("city", sa.String(100), nullable=True),
        sa.Column("state", sa.String(100), nullable=True),
        sa.Column("country", sa.String(2), nullable=True),
        sa.Column("tags", sa.String(500), nullable=True),
        sa.Column("custom", JSON, nullable=True, server_default=sa.text("'{}'::json")),
        sa.Column("created_by", UUID(as_uuid=True), sa.ForeignKey("users.id", ondelete="SET NULL"), nullable=True),
        sa.Column("updated_by", UUID(as_uuid=True), sa.ForeignKey("users.id", ondelete="SET NULL"), nullable=True),
        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
        sa.Column("deleted_at", sa.DateTime(timezone=True), nullable=True),
    )
    op.execute('CREATE INDEX IF NOT EXISTS ix_contactpersons_tenant_deleted ON contactpersons (tenant_id, deleted_at)')
    op.execute('CREATE INDEX IF NOT EXISTS ix_contactpersons_contact ON contactpersons (contact_id)')
    op.execute('CREATE INDEX IF NOT EXISTS ix_contactpersons_email ON contactpersons (email)')

    # ── 4. Add FK columns to contacts that reference contactpersons ───
    op.execute("ALTER TABLE contacts ADD COLUMN IF NOT EXISTS default_person_id UUID REFERENCES contactpersons(id) ON DELETE SET NULL")
    op.execute("ALTER TABLE contacts ADD COLUMN IF NOT EXISTS admin_contactperson_id UUID REFERENCES contactpersons(id) ON DELETE SET NULL")

    # ── 5. Migrate data from old tables ────────────────────────────────

    # 5a. companies_old → contacts (type='company')
    if _table_exists(conn, "companies_old"):
        # Build column list dynamically based on what exists in companies_old
        company_cols = {
            "id": "id",
            "tenant_id": "tenant_id",
            "name": "name",
            "phone": "phone_1",
            "email": "email_1",
            "website": "website",
            "description": "projectnote",
            "deleted_at": "deleted_at",
            "created_by": "created_by",
            "updated_by": "updated_by",
            "created_at": "created_at",
            "updated_at": "updated_at",
        }
        # account_number → code (check if it exists)
        if _column_exists(conn, "companies_old", "account_number"):
            company_cols["account_number"] = "code"
        # industry → tags (check if it exists)
        if _column_exists(conn, "companies_old", "industry"):
            company_cols["industry"] = "tags"

        select_cols = []
        insert_cols = []
        for old_col, new_col in company_cols.items():
            select_cols.append(old_col)
            insert_cols.append(new_col)

        # Build the INSERT ... SELECT statement
        select_list = ", ".join(f'"{c}"' for c in select_cols)
        # Add computed columns
        select_list += ", 'company' AS type, "
        # displayname = name
        if "name" in select_cols:
            select_list += '"name" AS displayname'
        else:
            select_list += "'' AS displayname"

        insert_list = ", ".join(f'"{c}"' for c in insert_cols) + ', "type", "displayname"'

        sql = f'INSERT INTO contacts ({insert_list}) SELECT {select_list} FROM companies_old'
        op.execute(sql)

        row_count = conn.execute(sa.text("SELECT COUNT(*) FROM companies_old")).scalar()
        logger.info("Migrated %d rows from companies_old → contacts (type='company')", row_count or 0)

    # 5b. contacts_old → contacts (type='person')
    if _table_exists(conn, "contacts_old"):
        # Map old contact columns to new contacts columns
        contact_cols = {
            "id": "id",
            "tenant_id": "tenant_id",
            "first_name": "firstname",
            "last_name": "surname",
            "email": "email_1",
            "phone": "phone_1",
            "deleted_at": "deleted_at",
            "created_by": "created_by",
            "updated_by": "updated_by",
            "created_at": "created_at",
            "updated_at": "updated_at",
        }
        # mobile → phone_2
        if _column_exists(conn, "contacts_old", "mobile"):
            contact_cols["mobile"] = "phone_2"
        # notes → projectnote
        if _column_exists(conn, "contacts_old", "notes"):
            contact_cols["notes"] = "projectnote"

        select_cols = []
        insert_cols = []
        for old_col, new_col in contact_cols.items():
            select_cols.append(old_col)
            insert_cols.append(new_col)

        select_list = ", ".join(f'"{c}"' for c in select_cols)
        # Add computed columns
        select_list += ", 'person' AS type, "
        # displayname = first_name || ' ' || last_name
        if _column_exists(conn, "contacts_old", "first_name") and _column_exists(conn, "contacts_old", "last_name"):
            select_list += "COALESCE(first_name, '') || ' ' || COALESCE(last_name, '') AS displayname"
        elif _column_exists(conn, "contacts_old", "first_name"):
            select_list += "first_name AS displayname"
        else:
            select_list += "'' AS displayname"

        insert_list = ", ".join(f'"{c}"' for c in insert_cols) + ', "type", "displayname"'

        sql = f'INSERT INTO contacts ({insert_list}) SELECT {select_list} FROM contacts_old'
        op.execute(sql)

        row_count = conn.execute(sa.text("SELECT COUNT(*) FROM contacts_old")).scalar()
        logger.info("Migrated %d rows from contacts_old → contacts (type='person')", row_count or 0)

    # 5c. company_contacts_old → contactpersons
    #     Each row links a company to a person. In the new schema, contactpersons
    #     are persons attached to a company contact. We map:
    #       contact_id (FK to contacts) = company_id (the company, now a contact)
    #       person details come from the old contacts table
    if _table_exists(conn, "company_contacts_old") and _table_exists(conn, "contacts_old"):
        sql = """
            INSERT INTO contactpersons (
                id, tenant_id, contact_id, displayname,
                firstname, lastname, function, phone, email,
                tags, created_at, updated_at, deleted_at
            )
            SELECT
                gen_random_uuid(),
                cc.tenant_id,
                cc.company_id,
                COALESCE(c.first_name, '') || ' ' || COALESCE(c.last_name, ''),
                c.first_name,
                c.last_name,
                cc.role_at_company,
                c.phone,
                c.email,
                CASE WHEN cc.is_primary THEN 'primary' ELSE NULL END,
                cc.created_at,
                cc.updated_at,
                cc.deleted_at
            FROM company_contacts_old cc
            JOIN contacts_old c ON cc.contact_id = c.id
        """
        op.execute(sql)

        row_count = conn.execute(sa.text("SELECT COUNT(*) FROM company_contacts_old")).scalar()
        logger.info("Migrated %d rows from company_contacts_old → contactpersons", row_count or 0)

    # ── 6. Enable RLS on new tables ───────────────────────────────────
    for table_name in ["contacts", "contactpersons"]:
        op.execute(f'ALTER TABLE "{table_name}" ENABLE ROW LEVEL SECURITY')
        op.execute(f'DROP POLICY IF EXISTS tenant_isolation ON "{table_name}"')
        op.execute(
            f'CREATE POLICY tenant_isolation ON "{table_name}" '
            f"USING (tenant_id = current_setting('app.current_tenant_id')::uuid)"
        )


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

    # Drop RLS policies on new tables
    for table_name in ["contactpersons", "contacts"]:
        op.execute(f'DROP POLICY IF EXISTS tenant_isolation ON "{table_name}"')
        op.execute(f'ALTER TABLE "{table_name}" DISABLE ROW LEVEL SECURITY')

    # Drop FK columns from contacts
    op.drop_column("contacts", "admin_contactperson_id")
    op.drop_column("contacts", "default_person_id")

    # Drop new tables
    op.drop_table("contactpersons")
    op.drop_table("contacts")

    # Restore old tables by renaming _old suffix back
    for tbl in ["companies", "contacts", "company_contacts"]:
        old_name = f"{tbl}_old"
        if _table_exists(conn, old_name) and not _table_exists(conn, tbl):
            op.execute(f'ALTER TABLE "{old_name}" RENAME TO "{tbl}"')
            logger.info("Restored %s → %s", old_name, tbl)
        elif _table_exists(conn, old_name) and _table_exists(conn, tbl):
            # Both exist — drop the _old version (new table takes precedence)
            op.execute(f'DROP TABLE "{old_name}" CASCADE')
            logger.info("Dropped leftover %s (new %s already exists)", old_name, tbl)