feat: sistema generico de database migrations com runner Python, auditoria e rollback

[mauriciomendonca]

Summary

Implementa um sistema genérico de migrations para o data-platform, substituindo a abordagem ad-hoc de scripts individuais por um runner Python único com auditoria, dry-run e rollback padronizados.

Motivação: O workflow db-migrate.yaml do PR #108 é funcional mas hardcoda o script migrate_unique_ids.py e validações específicas. Qualquer migration futura exigiria editar o workflow. Este PR resolve isso criando uma arquitetura reusável.

Closes #109

O que muda

1. Runner genérico (scripts/migrate.py)

• CLI com typer: status, migrate, rollback, history, validate
• Discovery automático de migrations em scripts/migrations/ por convenção de nome (NNN_desc.sql ou NNN_desc.py)
• Tabela migration_history com auditoria completa (quem, quando, duração, métricas JSONB)
• Bootstrap automático: importa entradas de schema_version na primeira execução
• Commit atômico: migration + registro de histórico na mesma transação
• --dry-run nativo em todos os comandos destrutivos
• Suporte a rollback via _rollback.sql (SQL) ou rollback() (Python)

2. Migration 006 (scripts/migrations/006_migrate_unique_ids.py)

• Adapta scripts/migrate_unique_ids.py para a interface do runner (describe/migrate/rollback)
• Extrai lógica de negócio (slugify, build_id_mapping, batch update) sem argparse ou conexão própria
• Script original marcado como DEPRECATED (preservado para backward compatibility com testes existentes)

3. Rollback SQL files

• 004_create_news_features_rollback.sql: remove tabela, trigger e índices
• 005_alter_unique_id_varchar_rollback.sql: reverte VARCHAR(120) → VARCHAR(32)

4. Schema updates

• docker/postgres/init.sql e scripts/create_schema.sql: adicionada tabela migration_history + view migration_status

5. Workflow generalizado (db-migrate.yaml)

• Inputs genéricos: command, dry_run, target_version, confirm
• Remove referências hardcoded a migrate_unique_ids.py e validações de unique_id
• Jobs: validate-inputs → backup → run-migration → show-status
• Cloud SQL Proxy version extraída para env
• Usa vars.GCP_WORKLOAD_IDENTITY_PROVIDER (variável de org)

6. Documentação

• docs/database/migrations.md: reescrito com documentação completa do runner
• docs/runbooks/migration-rollback.md: novo runbook com cenários A/B/C/D de rollback
• scripts/migrations/README.md: atualizado

Arquivos criados/modificados

Arquivo	Ação	Descrição
scripts/migrate.py	NOVO	Runner genérico (~400 linhas)
scripts/migrations/006_migrate_unique_ids.py	NOVO	Python migration adaptada
scripts/migrations/004_create_news_features_rollback.sql	NOVO	Rollback SQL
scripts/migrations/005_alter_unique_id_varchar_rollback.sql	NOVO	Rollback SQL
tests/unit/test_migrate_runner.py	NOVO	26 testes unitários
tests/unit/test_migration_006.py	NOVO	18 testes unitários
docs/runbooks/migration-rollback.md	NOVO	Runbook de rollback
.github/workflows/db-migrate.yaml	MODIFICADO	Generalizado
docker/postgres/init.sql	MODIFICADO	+migration_history
scripts/create_schema.sql	MODIFICADO	+migration_history
scripts/migrate_unique_ids.py	MODIFICADO	Marcado DEPRECATED
scripts/migrations/README.md	MODIFICADO	Atualizado
docs/database/migrations.md	REESCRITO	Nova documentação

Decisões técnicas

Decisão	Justificativa
Runner próprio vs Alembic	Projeto tem 6 migrations e ~300k registros. Script de ~400 linhas com psycopg2+typer é suficiente sem a complexidade de Alembic (revisions, env.py, alembic.ini)
migration_history vs schema_version	schema_version não registra quem executou, duração, status de falha, ou tipo de operação. Preservada para backward compatibility
Commit atômico	Migration + registro de histórico na mesma transação garante consistência
Discovery por convenção	Novas migrations são adicionadas apenas criando arquivo — sem editar runner ou workflow
importlib para Python migrations	Nomes com prefixo numérico (006_...) não são importáveis via import, necessitando importlib.util

Princípios seguidos

• SRP: Runner (orquestração) vs Migration 006 (lógica de negócio) vs CLI (interface)
• OCP: Novas migrations por convenção de arquivo, sem editar runner
• DIP: Funções recebem conn injetado, não criam conexão
• TDD: 44 testes escritos antes da implementação, ciclos RED→GREEN→REFACTOR

Backward-compatible

Este PR pode ser deployado a qualquer momento sem quebrar nada:

• Tabela migration_history criada com IF NOT EXISTS
• Bootstrap importa entradas existentes de schema_version
• Script original preservado como DEPRECATED
• Workflow aceita os mesmos comandos que antes (via status)

Dependências

• Depende de: feat: preparar schema, engagement e migração de dados para unique_id legível (Fases 2+3+4) #108 (já mergeado)
• Relacionada: Modificar Padrão de unique_id para Slug Legível #43 (issue original de unique_id legível)
• Nenhuma dependência externa nova (psycopg2, typer, loguru já no pyproject.toml)

Test plan

• 26 novos testes do runner: discovery, bootstrap, pending, execução SQL/Python, rollback, validate
• 18 novos testes da migration 006: interface, slugify, suffix, ID generation, migrate, rollback, dry-run, paridade
• Zero regressão: 255 testes existentes continuam passando (15 falhas pré-existentes não causadas por este PR)
• Verificação local: status → migrate --dry-run → migrate → history → rollback --dry-run → validate
• Verificação CI/CD: workflow db-migrate.yaml com command=status

# Rodar testes
PYTHONPATH=src poetry run pytest tests/unit/test_migrate_runner.py tests/unit/test_migration_006.py -v

# Verificação local (requer DATABASE_URL)
python scripts/migrate.py status
python scripts/migrate.py migrate --dry-run
python scripts/migrate.py validate

🤖 Generated with Claude Code

[mauriciomendonca]

Code Review — Correções Sugeridas

Revisão completa do PR. Quatro itens que devem ser corrigidos:

---

🔴 Problema #4 (ALTO): BEGIN/COMMIT explícito no rollback SQL quebra commit atômico

Arquivo: scripts/migrations/005_alter_unique_id_varchar_rollback.sql

O runner gerencia a transação (conn.autocommit = False). O BEGIN explícito é ignorado pelo PostgreSQL (já está em transação), mas o COMMIT commita prematuramente — antes que o runner registre o histórico em migration_history. Isso quebra a garantia de commit atômico.

Correção: Remover BEGIN; (linha 7) e COMMIT; (linha 47) do arquivo:

 -- 005_alter_unique_id_varchar_rollback.sql
 -- Rollback: revert unique_id to VARCHAR(32) and remove legacy_unique_id
 --
 -- WARNING: This rollback will FAIL if any unique_id exceeds 32 chars.
 -- Run 006 rollback first to restore MD5 IDs before running this.
-
-BEGIN;
 
 -- Step 1: Drop view that depends on news.unique_id
 DROP VIEW IF EXISTS news_with_themes;
 ...
 -- Step 5: Revert schema_version
 DELETE FROM schema_version WHERE version = '1.3';
-
-COMMIT;

---

🟡 Problema #1 (MÉDIO): fetchall() com server-side cursor anula o benefício

Arquivo: scripts/migrations/006_migrate_unique_ids.py:82-86

cursor = conn.cursor(name="fetch_news_for_migration")
cursor.itersize = 5000
cursor.execute(...)
rows = cursor.fetchall()  # ← carrega tudo em memória de qualquer forma

O name= cria server-side cursor, mas fetchall() puxa tudo de uma vez, anulando o benefício. Com ~300k rows isso é aceitável em memória, mas a semântica é confusa.

Correção: Remover o server-side cursor já que a intenção é carregar tudo:

 def _fetch_all_news(conn):
     """Fetch all news rows needed for migration."""
-    cursor = conn.cursor(name="fetch_news_for_migration")
-    cursor.itersize = 5000
+    cursor = conn.cursor()
     cursor.execute(
         "SELECT unique_id, agency_key, published_at, title, legacy_unique_id "
         "FROM news ORDER BY unique_id"
     )
     rows = cursor.fetchall()
     cursor.close()
     return rows

---

🟡 Problema #9 (MÉDIO): eval $CMD com input string livre no workflow

Arquivo: .github/workflows/db-migrate.yaml

O target_version é type: string livre. Um ator com permissão de workflow_dispatch poderia injetar comandos. Mesmo que quem tem write access já possa executar código, defense-in-depth sugere sanitizar.

Correção: Validar target_version no job validate-inputs e substituir eval por execução direta:

+      - name: Validate target_version format
+        if: inputs.target_version != ''
+        run: |
+          if ! echo "${{ inputs.target_version }}" | grep -qE '^[0-9]{3}$'; then
+            echo "::error::target_version must be a 3-digit number (e.g. 006)"
+            exit 1
+          fi
+
       - name: Check confirmation for destructive operations

E no step de execução, substituir eval $CMD por execução direta sem eval:

-          CMD="poetry run python scripts/migrate.py $COMMAND"
-
-          case "$COMMAND" in
-            status|history|validate)
-              CMD="$CMD --db-url \"$LOCAL_URL\""
-              ;;
-            migrate)
-              CMD="$CMD --db-url \"$LOCAL_URL\" --yes"
-              if [ "$DRY_RUN" = "true" ]; then
-                CMD="$CMD --dry-run"
-              fi
-              if [ -n "$TARGET" ]; then
-                CMD="$CMD --target $TARGET"
-              fi
-              ;;
-            rollback)
-              CMD="poetry run python scripts/migrate.py rollback $TARGET --db-url \"$LOCAL_URL\" --yes"
-              if [ "$DRY_RUN" = "true" ]; then
-                CMD="$CMD --dry-run"
-              fi
-              ;;
-          esac
-
-          echo "Executing: $COMMAND (dry_run=$DRY_RUN, target=$TARGET)"
-          eval $CMD
+          ARGS=()
+
+          case "$COMMAND" in
+            status|history|validate)
+              ARGS+=("$COMMAND" --db-url "$LOCAL_URL")
+              ;;
+            migrate)
+              ARGS+=("migrate" --db-url "$LOCAL_URL" --yes)
+              if [ "$DRY_RUN" = "true" ]; then
+                ARGS+=(--dry-run)
+              fi
+              if [ -n "$TARGET" ]; then
+                ARGS+=(--target "$TARGET")
+              fi
+              ;;
+            rollback)
+              ARGS+=("rollback" "$TARGET" --db-url "$LOCAL_URL" --yes)
+              if [ "$DRY_RUN" = "true" ]; then
+                ARGS+=(--dry-run)
+              fi
+              ;;
+          esac
+
+          echo "Executing: ${ARGS[*]}"
+          poetry run python scripts/migrate.py "${ARGS[@]}"

---

🟢 Problema #5 (BAIXO): Duplicação de lógica entre execute_migration e execute_rollback

Arquivo: scripts/migrate.py:293-423

As duas funções compartilham ~60% da lógica: logging, timing, record_history, error handling, dry_run rollback. A duplicação aumenta o risco de divergência futura.

Correção: Extrair um helper _execute_with_history que encapsula o padrão comum:

def _execute_with_history(
    conn,
    migration: MigrationInfo,
    operation: str,
    dry_run: bool,
    applied_by: str,
    run_id: str | None,
    action_fn,  # callable(conn) -> (description, execution_details)
):
    """Execute an action (migrate or rollback) with atomic history recording."""
    effective_op = "dry_run" if (dry_run and operation == "migrate") else operation
    started_at = time.time()

    logger.info(
        f"{'[DRY RUN] ' if dry_run else ''}"
        f"{'Rolling back' if operation == 'rollback' else 'Executing'} "
        f"{migration.version}_{migration.name} ({migration.migration_type})"
    )

    try:
        description, execution_details = action_fn(conn)

        if dry_run:
            _record_history(
                conn, migration, effective_op, "success", started_at,
                applied_by, run_id, description, execution_details,
            )
            conn.rollback()
            logger.info(f"[DRY RUN] {migration.version} previewed (rolled back)")
        else:
            _record_history(
                conn, migration, effective_op, "success", started_at,
                applied_by, run_id, description, execution_details,
            )
            conn.commit()
            logger.info(f"{migration.version}_{migration.name} {operation}d successfully")

    except Exception as e:
        conn.rollback()
        try:
            _record_history(
                conn, migration, effective_op, "failed", started_at,
                applied_by, run_id, error_message=str(e),
            )
            conn.commit()
        except Exception:
            logger.warning(f"Could not record {operation} failure in migration_history")
        raise

E simplificar as funções públicas:

def execute_migration(conn, migration, dry_run, applied_by, run_id):
    def action(conn):
        if migration.migration_type == "sql":
            cursor = conn.cursor()
            try:
                cursor.execute(migration.path.read_text())
            finally:
                cursor.close()
            return None, None
        else:
            module = _load_python_module(migration.path)
            if not hasattr(module, "describe"):
                raise AttributeError(f"Python migration {migration.path.name} must define describe()")
            description = module.describe()
            result = module.migrate(conn, dry_run=dry_run)
            return description, result if isinstance(result, dict) else None

    _execute_with_history(conn, migration, "migrate", dry_run, applied_by, run_id, action)


def execute_rollback(conn, migration, dry_run, applied_by, run_id):
    def action(conn):
        if migration.migration_type == "sql":
            if not migration.rollback_path or not migration.rollback_path.exists():
                raise FileNotFoundError(
                    f"No rollback file for SQL migration {migration.version}_{migration.name}"
                )
            cursor = conn.cursor()
            try:
                cursor.execute(migration.rollback_path.read_text())
            finally:
                cursor.close()
            return None, None
        else:
            module = _load_python_module(migration.path)
            result = module.rollback(conn, dry_run=dry_run)
            return None, result if isinstance(result, dict) else None

    _execute_with_history(conn, migration, "rollback", dry_run, applied_by, run_id, action)

Nota: O tratamento especial de NotImplementedError no rollback Python precisa ser preservado — pode ser tratado dentro do action com um early return ou no _execute_with_history como caso especial.

---

🤖 Review gerado com Claude Code