"""Carril CRM -> EFC: columnas efc_* en crm.cases + outbox transaccional (dos tablas).

El expediente NO se crea aquí: ya existe como ``crm.cases`` (ver d4e5f6a7b8c9), con su folio
en ``reference`` y su consecutivo en ``crm.folio_counters``. Esta migración solo le cuelga lo
que el carril hacia EFC necesita y crea el outbox. Es aditiva a propósito: la estructura del
expediente es de quien la definió, nosotros aportamos la conexión.

Las seis columnas ``efc_*`` son un ESPEJO de lo que hay en EFC, nunca el handle. El handle con
el que el CRM habla de un expediente es su ``id`` y su ``reference``: ``efc_pedimento_id`` es
un caché de la resolución y el ``pedimento_app`` del lado de EFC es mutable —se reescribe al
completar el provisional—, así que apoyarse en él rompería en cuanto llegue la data real.

``efc_storage_token`` es INMUTABLE una vez asignado: es la carpeta de MinIO donde EFC guarda
los objetos de este expediente. Que no cambie nunca es lo que permite completar el pedimento
sin mover un solo archivo.

``efc_link_state`` nace en 'PENDING' para las filas que ya existen. Eso es deliberado: al
encender el carril, el barrido de respaldo replica el histórico. Mientras ``EFC_API_URL`` esté
vacía el carril está apagado y no se encola nada.

Revision ID: c5d6e7f8a9b0
Revises: d4e5f6a7b8c9
Create Date: 2026-08-10 00:00:00.000000

"""
from typing import Sequence, Union

import sqlalchemy as sa
from alembic import op

revision: str = "c5d6e7f8a9b0"
down_revision: Union[str, None] = "d4e5f6a7b8c9"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None


def upgrade() -> None:
    # ---------- crm.cases: el espejo de EFC ----------
    op.add_column("cases", sa.Column("efc_organizacion_id", sa.String(length=36), nullable=True), schema="crm")
    op.add_column("cases", sa.Column("efc_pedimento_id", sa.String(length=36), nullable=True), schema="crm")
    op.add_column("cases", sa.Column("efc_storage_token", sa.String(length=25), nullable=True), schema="crm")
    op.add_column(
        "cases",
        # PENDING | LINKED | FAILED
        sa.Column("efc_link_state", sa.String(length=20), nullable=False, server_default=sa.text("'PENDING'")),
        schema="crm",
    )
    # El diagnóstico se guarda en la fila para que se vea en la ficha del expediente, sin
    # obligar a nadie a ir a los logs del worker.
    op.add_column("cases", sa.Column("efc_error_code", sa.String(length=60), nullable=True), schema="crm")
    op.add_column("cases", sa.Column("efc_error_detail", sa.Text(), nullable=True), schema="crm")
    op.create_index("ix_crm_cases_efc_link_state", "cases", ["efc_link_state"], schema="crm")

    # Relleno del token para los expedientes que ya existían. Sin esto, el barrido de
    # reconciliación los encola, EFC los rechaza con
    # {'storage_token': ['This field may not be null.']} y agotan sus 8 intentos hasta
    # quedar en `failed`: ruido permanente por un dato que se podía derivar.
    #
    # La condición de longitud replica la guarda de storage_token(): en los 25 caracteres de
    # `pedimento_app` caben hasta 6 dígitos de company. Lo que no cabe se queda NULL a
    # propósito y el encolado lo salta avisando, porque un token recortado apuntaría a la
    # carpeta de otro expediente y mezclaría documentos en silencio.
    #
    # `reference IS NOT NULL` porque el folio es nullable en crm.cases: un expediente sin
    # folio no tiene con qué identificarse ante EFC.
    op.execute(
        """
        UPDATE crm.cases
           SET efc_storage_token = 'CRM-' || company_id::text || '-' || reference
         WHERE efc_storage_token IS NULL
           AND reference IS NOT NULL
           AND length('CRM-' || company_id::text || '-' || reference) <= 25
        """
    )

    # ---------- crm.efc_sync_outbox: metadatos (alta del provisional y completado) ----------
    op.create_table(
        "efc_sync_outbox",
        sa.Column("id", sa.Integer(), nullable=False, autoincrement=True),
        sa.Column("kind", sa.String(length=20), nullable=False),
        sa.Column("payload", sa.JSON(), nullable=False),
        sa.Column("expediente_ref", sa.Integer(), nullable=True),
        sa.Column("status", sa.String(length=10), nullable=False, server_default=sa.text("'pending'")),
        sa.Column("attempts", sa.Integer(), nullable=False, server_default=sa.text("0")),
        sa.Column("last_error", sa.Text(), nullable=True),
        sa.Column("sent_at", sa.DateTime(), nullable=True),
        sa.Column("efc_pedimento_id", sa.String(length=36), nullable=True),
        sa.Column("tenant_id", sa.Integer(), nullable=False),
        sa.Column("company_id", sa.Integer(), nullable=False),
        sa.Column("created_at", sa.DateTime(), nullable=False, server_default=sa.text("now()")),
        sa.Column("updated_at", sa.DateTime(), nullable=False, server_default=sa.text("now()")),
        sa.Column("deleted_at", sa.DateTime(), nullable=True),
        sa.PrimaryKeyConstraint("id"),
        schema="crm",
    )
    op.create_index("ix_crm_efc_sync_outbox_status", "efc_sync_outbox", ["status"], schema="crm")
    op.create_index("ix_crm_efc_sync_outbox_kind_status", "efc_sync_outbox", ["kind", "status"], schema="crm")
    op.create_index("ix_crm_efc_sync_outbox_expediente_ref", "efc_sync_outbox", ["expediente_ref"], schema="crm")
    op.create_index("ix_crm_efc_sync_outbox_tenant_id", "efc_sync_outbox", ["tenant_id"], schema="crm")
    op.create_index("ix_crm_efc_sync_outbox_company_id", "efc_sync_outbox", ["company_id"], schema="crm")
    op.create_foreign_key(
        "fk_crm_efc_sync_outbox_tenant_id", "efc_sync_outbox", "tenants",
        ["tenant_id"], ["id"], source_schema="crm", referent_schema="core",
    )

    # ---------- crm.efc_file_outbox: archivos ----------
    op.create_table(
        "efc_file_outbox",
        sa.Column("id", sa.Integer(), nullable=False, autoincrement=True),
        sa.Column("kind", sa.String(length=30), nullable=False),
        sa.Column("s3_key", sa.String(length=1024), nullable=False),
        sa.Column("file_name", sa.String(length=255), nullable=False),
        sa.Column("content_type", sa.String(length=100), nullable=True),
        sa.Column("efc_tipo", sa.String(length=40), nullable=False),
        # La pareja (tabla, id) desambigua entre las DOS secuencias de documentos del CRM:
        # crm.documents.id = 5 y ops.shipment_documents.id = 5 coexisten.
        sa.Column("source_table", sa.String(length=30), nullable=False),
        sa.Column("source_id", sa.Integer(), nullable=True),
        sa.Column("crm_document_ref", sa.String(length=64), nullable=True),
        sa.Column("expediente_ref", sa.Integer(), nullable=False),
        sa.Column("delete_local", sa.Boolean(), nullable=False, server_default=sa.text("true")),
        sa.Column("status", sa.String(length=10), nullable=False, server_default=sa.text("'pending'")),
        sa.Column("attempts", sa.Integer(), nullable=False, server_default=sa.text("0")),
        sa.Column("last_error", sa.Text(), nullable=True),
        sa.Column("sent_at", sa.DateTime(), nullable=True),
        sa.Column("efc_document_id", sa.String(length=36), nullable=True),
        sa.Column("tenant_id", sa.Integer(), nullable=False),
        sa.Column("company_id", sa.Integer(), nullable=False),
        sa.Column("created_at", sa.DateTime(), nullable=False, server_default=sa.text("now()")),
        sa.Column("updated_at", sa.DateTime(), nullable=False, server_default=sa.text("now()")),
        sa.Column("deleted_at", sa.DateTime(), nullable=True),
        sa.PrimaryKeyConstraint("id"),
        schema="crm",
    )
    op.create_index("ix_crm_efc_file_outbox_status", "efc_file_outbox", ["status"], schema="crm")
    op.create_index("ix_crm_efc_file_outbox_kind_status", "efc_file_outbox", ["kind", "status"], schema="crm")
    # Índice de la guarda _ya_entregado, que es lo que se consulta en cada encolado.
    op.create_index("ix_crm_efc_file_outbox_source", "efc_file_outbox", ["source_table", "source_id"], schema="crm")
    op.create_index("ix_crm_efc_file_outbox_expediente_ref", "efc_file_outbox", ["expediente_ref"], schema="crm")
    op.create_index("ix_crm_efc_file_outbox_tenant_id", "efc_file_outbox", ["tenant_id"], schema="crm")
    op.create_index("ix_crm_efc_file_outbox_company_id", "efc_file_outbox", ["company_id"], schema="crm")
    op.create_foreign_key(
        "fk_crm_efc_file_outbox_tenant_id", "efc_file_outbox", "tenants",
        ["tenant_id"], ["id"], source_schema="crm", referent_schema="core",
    )
    # expediente_ref -> crm.cases.id. El nombre de la columna conserva el término del dominio:
    # crm.cases ES el expediente (así lo nombra su propio docstring), y el carril, EFC y el
    # ticket hablan de expedientes. Renombrarlo a case_ref solo movería la ambigüedad de sitio.
    op.create_foreign_key(
        "fk_crm_efc_file_outbox_expediente_ref", "efc_file_outbox", "cases",
        ["expediente_ref"], ["id"], source_schema="crm", referent_schema="crm",
    )


def downgrade() -> None:
    op.drop_constraint("fk_crm_efc_file_outbox_expediente_ref", "efc_file_outbox", schema="crm", type_="foreignkey")
    op.drop_constraint("fk_crm_efc_file_outbox_tenant_id", "efc_file_outbox", schema="crm", type_="foreignkey")
    op.drop_index("ix_crm_efc_file_outbox_company_id", table_name="efc_file_outbox", schema="crm")
    op.drop_index("ix_crm_efc_file_outbox_tenant_id", table_name="efc_file_outbox", schema="crm")
    op.drop_index("ix_crm_efc_file_outbox_expediente_ref", table_name="efc_file_outbox", schema="crm")
    op.drop_index("ix_crm_efc_file_outbox_source", table_name="efc_file_outbox", schema="crm")
    op.drop_index("ix_crm_efc_file_outbox_kind_status", table_name="efc_file_outbox", schema="crm")
    op.drop_index("ix_crm_efc_file_outbox_status", table_name="efc_file_outbox", schema="crm")
    op.drop_table("efc_file_outbox", schema="crm")

    op.drop_constraint("fk_crm_efc_sync_outbox_tenant_id", "efc_sync_outbox", schema="crm", type_="foreignkey")
    op.drop_index("ix_crm_efc_sync_outbox_company_id", table_name="efc_sync_outbox", schema="crm")
    op.drop_index("ix_crm_efc_sync_outbox_tenant_id", table_name="efc_sync_outbox", schema="crm")
    op.drop_index("ix_crm_efc_sync_outbox_expediente_ref", table_name="efc_sync_outbox", schema="crm")
    op.drop_index("ix_crm_efc_sync_outbox_kind_status", table_name="efc_sync_outbox", schema="crm")
    op.drop_index("ix_crm_efc_sync_outbox_status", table_name="efc_sync_outbox", schema="crm")
    op.drop_table("efc_sync_outbox", schema="crm")

    op.drop_index("ix_crm_cases_efc_link_state", table_name="cases", schema="crm")
    op.drop_column("cases", "efc_error_detail", schema="crm")
    op.drop_column("cases", "efc_error_code", schema="crm")
    op.drop_column("cases", "efc_link_state", schema="crm")
    op.drop_column("cases", "efc_storage_token", schema="crm")
    op.drop_column("cases", "efc_pedimento_id", schema="crm")
    op.drop_column("cases", "efc_organizacion_id", schema="crm")
