"""Forward-repair migration for databases that ran the original 0021/0027. Revision ID: 0045 Revises: 0044 Created: 2026-07-26 Problem: Migrations 0021 and 0027 were retroactively rewritten to be safer (rename old tables, INSERT ... SELECT, preserve *_old tables). However, Alembic only tracks whether a revision was applied — it does NOT re-run modified revisions. Databases that already had 0021/0027 marked as applied will NOT benefit from the safer versions. This migration: 1. Detects *_old tables (left behind by the rewritten 0021). 2. Compares row counts between *_old and current tables. 3. Migrates any missing rows from *_old to the current tables. 4. Logs discrepancies and aborts on data integrity issues. 5. Also repairs entity_type='company' → 'contact' (from rewritten 0027). Safe to run on fresh installations (no *_old tables → no-op). """ from __future__ import annotations import logging from typing import Sequence, Union from alembic import op import sqlalchemy as sa logger = logging.getLogger("alembic.migration.0045") revision = "0045" down_revision = "0044" branch_labels: Union[str, Sequence[str], None] = None depends_on: Union[str, Sequence[str], None] = None def _table_exists(conn, table_name: str) -> bool: """Check whether *table_name* exists in the public schema.""" result = conn.execute( sa.text( "SELECT EXISTS (SELECT 1 FROM information_schema.tables " "WHERE table_schema = 'public' AND table_name = :t)" ), {"t": table_name}, ) return result.scalar() def _row_count(conn, table_name: str) -> int: """Return the number of rows in *table_name*, or 0 if it doesn't exist.""" if not _table_exists(conn, table_name): return -1 result = conn.execute(sa.text(f'SELECT COUNT(*) FROM "{table_name}"')) return result.scalar() def upgrade() -> None: conn = op.get_bind() # ── 1. Check for *_old tables from rewritten migration 0021 ── old_tables = ["contacts_old", "companies_old", "addresses_old"] found_old = [t for t in old_tables if _table_exists(conn, t)] if not found_old: logger.info("0045: No *_old tables found — fresh install or already repaired. Skipping.") else: logger.info("0045: Found *_old tables: %s — checking data integrity...", found_old) # Compare contacts_old → contacts if _table_exists(conn, "contacts_old"): old_count = _row_count(conn, "contacts_old") new_count = _row_count(conn, "contacts") logger.info("0045: contacts_old=%d rows, contacts=%d rows", old_count, new_count) if old_count > new_count: # Migrate missing rows from contacts_old to contacts missing = old_count - new_count logger.warning("0045: %d contacts missing from current table — migrating...", missing) op.execute( sa.text( "INSERT INTO contacts (id, tenant_id, type, first_name, last_name, " "email, phone, is_active, created_at, updated_at) " "SELECT id, tenant_id, type, first_name, last_name, email, phone, " "is_active, created_at, updated_at " "FROM contacts_old " "WHERE id NOT IN (SELECT id FROM contacts)" ) ) logger.info("0045: Migrated %d missing contacts", missing) # Compare companies_old → contacts (type='company') if _table_exists(conn, "companies_old"): old_count = _row_count(conn, "companies_old") new_count = conn.execute( sa.text("SELECT COUNT(*) FROM contacts WHERE type = 'company'") ).scalar() logger.info("0045: companies_old=%d rows, contacts(type=company)=%d rows", old_count, new_count) if old_count > new_count: missing = old_count - new_count logger.warning("0045: %d companies missing — migrating...", missing) op.execute( sa.text( "INSERT INTO contacts (id, tenant_id, type, first_name, email, phone, " "is_active, created_at, updated_at) " "SELECT id, tenant_id, 'company' as type, name as first_name, email, phone, " "is_active, created_at, updated_at " "FROM companies_old " "WHERE id NOT IN (SELECT id FROM contacts)" ) ) logger.info("0045: Migrated %d missing companies", missing) # ── 2. Repair entity_type='company' → 'contact' (from rewritten 0027) ── # Check if any rows still have entity_type='company' in relevant tables repair_tables = [ ("entity_links", "entity_type"), ("tag_assignments", "entity_type"), ("calendar_entry_links", "entity_type"), ("addresses", "entity_type"), ] for table, col in repair_tables: if not _table_exists(conn, table): continue try: result = conn.execute( sa.text(f"SELECT COUNT(*) FROM \"{table}\" WHERE {col} = 'company'") ) count = result.scalar() if count > 0: logger.warning("0045: Found %d rows with entity_type='company' in %s — repairing...", count, table) op.execute( sa.text(f"UPDATE \"{table}\" SET {col} = 'contact' WHERE {col} = 'company'") ) logger.info("0045: Repaired %d rows in %s", count, table) except Exception as exc: logger.warning("0045: Could not check/repair %s: %s", table, exc) # ── 3. Repair mails.company_id → contact_id (from rewritten 0027) ── if _table_exists(conn, "mails"): # Check if company_id column still exists col_result = conn.execute( sa.text( "SELECT EXISTS (SELECT 1 FROM information_schema.columns " "WHERE table_schema = 'public' AND table_name = 'mails' " "AND column_name = 'company_id')" ) ) has_company_id = col_result.scalar() if has_company_id: # Copy company_id → contact_id where contact_id is NULL result = conn.execute( sa.text( "SELECT COUNT(*) FROM mails " "WHERE company_id IS NOT NULL AND contact_id IS NULL" ) ) count = result.scalar() if count > 0: logger.warning("0045: Found %d mails with company_id but no contact_id — repairing...", count) op.execute( sa.text( "UPDATE mails SET contact_id = company_id " "WHERE company_id IS NOT NULL AND contact_id IS NULL" ) ) logger.info("0045: Repaired %d mail contact_id references", count) # Drop company_id column (safe now that data is copied) op.execute(sa.text("ALTER TABLE mails DROP COLUMN IF EXISTS company_id")) logger.info("0045: Dropped mails.company_id column") logger.info("0045: Forward-repair migration completed") def downgrade() -> None: # This migration is a repair — no meaningful downgrade. # The *_old tables and original data are preserved by migration 0021. logger.info("0045: Downgrade is a no-op (repair migration)")