"""partial unique indexes for soft-deleted rows

Revision ID: 21ca6cb7ae17
Revises: 2458d88ae1f9
Create Date: 2026-08-04 16:22:11.462457+06:00

"""
from typing import Sequence, Union

from alembic import op
import sqlalchemy as sa


revision: str = '21ca6cb7ae17'
down_revision: Union[str, None] = '2458d88ae1f9'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None


def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_index('ix_complaints_ticket_no', table_name='complaints')
    op.create_index(op.f('ix_complaints_ticket_no'), 'complaints', ['ticket_no'], unique=False)
    op.create_index('uq_complaints_ticket_no_active', 'complaints', ['ticket_no'], unique=True, postgresql_where=sa.text('is_deleted = false'))
    op.drop_index('ix_connection_applications_application_no', table_name='connection_applications')
    op.create_index(op.f('ix_connection_applications_application_no'), 'connection_applications', ['application_no'], unique=False)
    op.create_index('uq_applications_application_no_active', 'connection_applications', ['application_no'], unique=True, postgresql_where=sa.text('is_deleted = false'))
    op.drop_constraint('uq_connections_application_id', 'connections', type_='unique')
    op.drop_index('ix_connections_connection_code', table_name='connections')
    op.create_index(op.f('ix_connections_connection_code'), 'connections', ['connection_code'], unique=False)
    op.create_index('uq_connections_application_id_active', 'connections', ['application_id'], unique=True, postgresql_where=sa.text('is_deleted = false AND application_id IS NOT NULL'))
    op.create_index('uq_connections_connection_code_active', 'connections', ['connection_code'], unique=True, postgresql_where=sa.text('is_deleted = false'))
    op.drop_index('ix_customers_customer_code', table_name='customers')
    op.create_index(op.f('ix_customers_customer_code'), 'customers', ['customer_code'], unique=False)
    op.create_index('uq_customers_customer_code_active', 'customers', ['customer_code'], unique=True, postgresql_where=sa.text('is_deleted = false'))
    op.drop_index('ix_equipment_items_serial_number', table_name='equipment_items')
    op.create_index(op.f('ix_equipment_items_serial_number'), 'equipment_items', ['serial_number'], unique=False)
    op.create_index('uq_equipment_items_serial_active', 'equipment_items', ['serial_number'], unique=True, postgresql_where=sa.text('is_deleted = false AND serial_number IS NOT NULL'))
    op.drop_constraint('uq_users_employee_code', 'users', type_='unique')
    op.drop_index('ix_users_email', table_name='users')
    op.create_index(op.f('ix_users_email'), 'users', ['email'], unique=False)
    op.create_index('uq_users_email_active', 'users', ['email'], unique=True, postgresql_where=sa.text('is_deleted = false'))
    op.create_index('uq_users_employee_code_active', 'users', ['employee_code'], unique=True, postgresql_where=sa.text('is_deleted = false AND employee_code IS NOT NULL'))
    # ### end Alembic commands ###


def _assert_no_soft_deleted_duplicates() -> None:
    """Block the downgrade when live data relies on partial uniqueness.

    Reverting restores FULL uniqueness, but this revision exists precisely so a
    soft-deleted row can free its natural key for reuse. If any duplicates exist,
    recreating the plain UNIQUE index fails with a bare UniqueViolation that gives
    the operator nothing to act on — so check first and say what to fix.
    """
    conn = op.get_bind()
    checks = [
        ("users", "email"),
        ("users", "employee_code"),
        ("connection_applications", "application_no"),
        ("complaints", "ticket_no"),
        ("customers", "customer_code"),
        ("connections", "connection_code"),
        ("connections", "application_id"),
        ("equipment_items", "serial_number"),
    ]
    conflicts: list[str] = []
    for table, column in checks:
        dupes = conn.execute(
            sa.text(
                f"SELECT {column}, count(*) AS n FROM {table} "
                f"WHERE {column} IS NOT NULL GROUP BY {column} HAVING count(*) > 1"
            )
        ).fetchall()
        for value, count in dupes:
            conflicts.append(f"  {table}.{column} = {value!r} ({count} rows)")

    if conflicts:
        raise RuntimeError(
            "Cannot downgrade: restoring full UNIQUE constraints would fail because "
            "soft-deleted rows share a natural key with live rows.\n"
            + "\n".join(conflicts)
            + "\n\nResolve by hard-deleting the obsolete rows (e.g. "
            "DELETE FROM users WHERE is_deleted = true) or by renaming their keys, "
            "then re-run the downgrade."
        )


def downgrade() -> None:
    _assert_no_soft_deleted_duplicates()
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_index('uq_users_employee_code_active', table_name='users', postgresql_where=sa.text('is_deleted = false AND employee_code IS NOT NULL'))
    op.drop_index('uq_users_email_active', table_name='users', postgresql_where=sa.text('is_deleted = false'))
    op.drop_index(op.f('ix_users_email'), table_name='users')
    op.create_index('ix_users_email', 'users', ['email'], unique=True)
    op.create_unique_constraint('uq_users_employee_code', 'users', ['employee_code'])
    op.drop_index('uq_equipment_items_serial_active', table_name='equipment_items', postgresql_where=sa.text('is_deleted = false AND serial_number IS NOT NULL'))
    op.drop_index(op.f('ix_equipment_items_serial_number'), table_name='equipment_items')
    op.create_index('ix_equipment_items_serial_number', 'equipment_items', ['serial_number'], unique=True)
    op.drop_index('uq_customers_customer_code_active', table_name='customers', postgresql_where=sa.text('is_deleted = false'))
    op.drop_index(op.f('ix_customers_customer_code'), table_name='customers')
    op.create_index('ix_customers_customer_code', 'customers', ['customer_code'], unique=True)
    op.drop_index('uq_connections_connection_code_active', table_name='connections', postgresql_where=sa.text('is_deleted = false'))
    op.drop_index('uq_connections_application_id_active', table_name='connections', postgresql_where=sa.text('is_deleted = false AND application_id IS NOT NULL'))
    op.drop_index(op.f('ix_connections_connection_code'), table_name='connections')
    op.create_index('ix_connections_connection_code', 'connections', ['connection_code'], unique=True)
    op.create_unique_constraint('uq_connections_application_id', 'connections', ['application_id'])
    op.drop_index('uq_applications_application_no_active', table_name='connection_applications', postgresql_where=sa.text('is_deleted = false'))
    op.drop_index(op.f('ix_connection_applications_application_no'), table_name='connection_applications')
    op.create_index('ix_connection_applications_application_no', 'connection_applications', ['application_no'], unique=True)
    op.drop_index('uq_complaints_ticket_no_active', table_name='complaints', postgresql_where=sa.text('is_deleted = false'))
    op.drop_index(op.f('ix_complaints_ticket_no'), table_name='complaints')
    op.create_index('ix_complaints_ticket_no', 'complaints', ['ticket_no'], unique=True)
    # ### end Alembic commands ###